Merge pull request #10 from maxipavlovic/feature/LITE-17018

LITE-17018 Commands for check and sync of deleted objects in Replica

Author: d3rky

README.md

@@ -146,6 +146,13 @@ kubectl exec -i MASTER_CONTAINER -- python manage.py cqrs_diff_master --cqrs-id=
     kubectl exec -i MASTER_CONTAINER -- python manage.py cqrs_diff_sync
 ```
 
+* If it's important to check sync and clean up deleted objects within replica service in K8S:
+```bash
+kubectl exec -i REPLICA_CONTAINER -- python manage.py cqrs_deleted_diff_replica --cqrs-id=author | 
+    kubectl exec -i MASTER_CONTAINER -- python manage.py cqrs_deleted_diff_master |
+    kubectl exec -i REPLICA_CONTAINER -- python manage.py cqrs_deleted_sync_replica
+```
+
 Development
 ===========
 

dj_cqrs/management/commands/cqrs_deleted_diff_master.py

@@ -0,0 +1,68 @@
+#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
+
+import sys
+
+import ujson
+
+from django.core.management.base import BaseCommand, CommandError
+from django.db import connection
+
+from dj_cqrs.registries import MasterRegistry
+
+
+GET_NON_EXISTING_PKS_SQL_TEMPLATE = """
+SELECT t.pk
+FROM (
+     WITH t0(pk) AS (
+         VALUES {values}
+     )
+     SELECT *
+     FROM t0
+ ) t
+LEFT JOIN {table} m ON m.{pk_field} = t.pk
+WHERE m.{pk_field} IS NULL
+"""
+
+
+class Command(BaseCommand):
+    help = 'Diff of deleted CQRS models pks from master diff stream.'
+
+    @classmethod
+    def serialize_out(cls, package):
+        return ujson.dumps(package)
+
+    @classmethod
+    def deserialize_in(cls, package_line):
+        return ujson.loads(package_line)
+
+    def handle(self, *args, **options):
+        with sys.stdin as f:
+            first_line = f.readline()
+            model = self._get_model(first_line)
+            self.stdout.write(first_line.strip())
+
+            with connection.cursor() as cursor:
+                for package_line in f:
+                    master_data = self.deserialize_in(package_line)
+
+                    sql = GET_NON_EXISTING_PKS_SQL_TEMPLATE.format(
+                        values=','.join(["({})".format(pk) for pk in master_data]),
+                        table=model._meta.db_table,
+                        pk_field=model._meta.pk.attname,
+                    )
+
+                    cursor.execute(sql)
+                    diff_ids = [r[0] for r in cursor.fetchall()]
+                    if diff_ids:
+                        self.stdout.write(self.serialize_out(diff_ids))
+                        self.stderr.write('PK to delete: {}'.format(str(diff_ids)))
+
+    @staticmethod
+    def _get_model(first_line):
+        cqrs_id = first_line.split(',')[0]
+        model = MasterRegistry.get_model_by_cqrs_id(cqrs_id)
+
+        if not model:
+            raise CommandError('Wrong CQRS ID: {}!'.format(cqrs_id))
+
+        return model

dj_cqrs/management/commands/cqrs_deleted_diff_replica.py

@@ -0,0 +1,79 @@
+#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
+
+import ujson
+from django.core.exceptions import FieldError
+from django.core.management.base import BaseCommand, CommandError
+from django.utils.timezone import now
+
+from dj_cqrs.management.commands.utils import batch_qs
+from dj_cqrs.registries import ReplicaRegistry
+
+
+class Command(BaseCommand):
+    help = 'Streaming diff of CQRS model pks from replica service to check for deleted objects.'
+
+    @classmethod
+    def serialize_package(cls, package):
+        return ujson.dumps(package)
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--cqrs-id', '-cid',
+            help='CQRS_ID of the replica model',
+            type=str,
+            required=True,
+        )
+        parser.add_argument(
+            '--filter', '-f',
+            help='Filter kwargs',
+            type=str,
+            default=None,
+        )
+        parser.add_argument(
+            '--batch', '-b',
+            help='Batch size',
+            type=int,
+            default=10000,
+        )
+
+    def handle(self, *args, **options):
+        model = self._get_model(options)
+        batch_size = self._get_batch_size(options)
+
+        qs = model._default_manager.values().order_by()
+        if options['filter']:
+            try:
+                kwargs = ujson.loads(options['filter'])
+                if not isinstance(kwargs, dict):
+                    raise ValueError
+            except ValueError:
+                raise CommandError('Bad filter kwargs!')
+
+            try:
+                qs = qs.filter(**kwargs)
+            except FieldError as e:
+                raise CommandError('Bad filter kwargs! {}'.format(str(e)))
+
+        if not qs.exists():
+            self.stderr.write('No objects found for filter!')
+            return
+
+        current_dt = now()
+        self.stdout.write('{},{}'.format(model.CQRS_ID, str(current_dt)))
+
+        for bqs in batch_qs(qs.values_list('pk', flat=True), batch_size=batch_size):
+            self.stdout.write(self.serialize_package(list(bqs)))
+
+    @staticmethod
+    def _get_model(options):
+        cqrs_id = options['cqrs_id']
+        model = ReplicaRegistry.get_model_by_cqrs_id(cqrs_id)
+
+        if not model:
+            raise CommandError('Wrong CQRS ID: {}!'.format(cqrs_id))
+
+        return model
+
+    @staticmethod
+    def _get_batch_size(options):
+        return options['batch']

dj_cqrs/management/commands/cqrs_deleted_sync_replica.py

@@ -0,0 +1,41 @@
+#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
+
+import sys
+
+import ujson
+
+from django.core.management.base import BaseCommand, CommandError
+from django.db import DatabaseError
+
+from dj_cqrs.registries import ReplicaRegistry
+
+
+class Command(BaseCommand):
+    help = 'Diff for deleted objects synchronizer from CQRS master stream.'
+
+    @classmethod
+    def deserialize_in(cls, package_line):
+        return ujson.loads(package_line)
+
+    def handle(self, *args, **options):
+        with sys.stdin as f:
+            first_line = f.readline().strip()
+            model = self._get_model(first_line)
+
+            for pks_line in f:
+                try:
+                    model._default_manager.filter(
+                        pk__in=self.deserialize_in(pks_line.strip()),
+                    ).delete()
+                except DatabaseError as e:
+                    print(str(e), file=sys.stderr)
+
+    @staticmethod
+    def _get_model(first_line):
+        cqrs_id = first_line.split(',')[0]
+        model = ReplicaRegistry.get_model_by_cqrs_id(cqrs_id)
+
+        if not model:
+            raise CommandError('Wrong CQRS ID: {}!'.format(cqrs_id))
+
+        return model

tests/test_commands/test_deleted_diff_master.py

@@ -0,0 +1,110 @@
+#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
+
+import sys
+from io import StringIO
+
+import pytest
+from django.core.management import CommandError, call_command
+from django.utils.timezone import now
+
+from tests.dj_master.models import Author
+from tests.dj_replica.models import AuthorRef
+
+
+COMMAND_NAME = 'cqrs_deleted_diff_master'
+
+
+def test_bad_cqrs_id(mocker):
+    mocker.patch.object(sys, 'stdin', StringIO('invalid,datetime\n'))
+
+    with pytest.raises(CommandError) as e:
+        call_command(COMMAND_NAME)
+
+    assert 'Wrong CQRS ID: invalid!' in str(e)
+
+
+@pytest.mark.django_db
+def test_no_rows(mocker, capsys):
+    mocker.patch.object(sys, 'stdin', StringIO('author,datetime\n'))
+    call_command(COMMAND_NAME)
+
+    captured = capsys.readouterr()
+    assert not captured.err
+    assert 'author,datetime\n' == captured.out
+
+
+def replica_master_pipe(capsys, mocker, *args):
+    call_command('cqrs_deleted_diff_replica', '--cqrs-id=author', *args)
+    captured = capsys.readouterr()
+    mocker.patch.object(sys, 'stdin', StringIO(captured.out))
+    call_command(COMMAND_NAME)
+    return capsys.readouterr()
+
+
+@pytest.mark.django_db
+def test_first_row(capsys, mocker):
+    AuthorRef.objects.create(name='author', id=1, cqrs_revision=0, cqrs_updated=now())
+
+    captured = replica_master_pipe(capsys, mocker)
+
+    first_row = captured.out.split('\n')[0]
+    assert '{},'.format(Author.CQRS_ID) in first_row
+
+
+@pytest.mark.django_db
+def test_all_synced(capsys, mocker):
+    mocker.patch('dj_cqrs.controller.producer.produce')
+
+    Author.objects.create(id=1, name='name')
+    AuthorRef.objects.create(id=1, name='name', cqrs_revision=0, cqrs_updated=now())
+
+    captured = replica_master_pipe(capsys, mocker)
+    assert not captured.err
+
+
+@pytest.mark.django_db
+def test_sync_for_all(mocker, capsys):
+    for i in range(3):
+        AuthorRef.objects.create(name=str(i), id=i, cqrs_revision=i, cqrs_updated=now())
+
+    captured = replica_master_pipe(capsys, mocker)
+    for err_text in ('PK to delete', '0', '1', '2'):
+        assert err_text in captured.err
+
+    second_row = captured.out.split('\n')[1]
+    assert '[0,1,2]' in second_row
+
+
+@pytest.mark.django_db
+def test_partial_sync(mocker, capsys):
+    mocker.patch('dj_cqrs.controller.producer.produce')
+
+    Author.objects.create(id=1, name='name')
+    AuthorRef.objects.create(id=1, name='name', cqrs_revision=0, cqrs_updated=now())
+    AuthorRef.objects.create(id=2, name='2', cqrs_revision=0, cqrs_updated=now())
+
+    captured = replica_master_pipe(capsys, mocker)
+    assert 'PK to delete' in captured.err
+    assert '2' in captured.err
+
+    second_row = captured.out.split('\n')[1]
+    assert '[2]' in second_row
+
+
+@pytest.mark.django_db
+def test_sync_batch(mocker, capsys):
+    mocker.patch('dj_cqrs.controller.producer.produce')
+
+    for i in range(3):
+        AuthorRef.objects.create(id=i, name=str(i), cqrs_revision=0, cqrs_updated=now())
+    Author.objects.create(id=1, name='name')
+
+    captured = replica_master_pipe(capsys, mocker, '--batch=1')
+    for err_text in ('PK to delete', '0', '2'):
+        assert err_text in captured.err
+
+    second_row = captured.out.split('\n')[1]
+    assert '[0]' in second_row
+
+    third_row = captured.out.split('\n')[2]
+    assert '[2]' in third_row

tests/test_commands/test_deleted_diff_replica.py

@@ -0,0 +1,80 @@
+#  Copyright © 2021 Ingram Micro Inc. All rights reserved.
+
+import ujson
+
+import pytest
+from django.core.management import CommandError, call_command
+from django.utils.timezone import now
+
+from tests.dj_replica.models import AuthorRef
+
+
+COMMAND_NAME = 'cqrs_deleted_diff_replica'
+
+
+def test_no_cqrs_id():
+    with pytest.raises(CommandError):
+        call_command(COMMAND_NAME)
+
+
+def test_bad_cqrs_id():
+    with pytest.raises(CommandError) as e:
+        call_command(COMMAND_NAME, '--cqrs-id=invalid')
+
+    assert 'Wrong CQRS ID: invalid!' in str(e)
+
+
+@pytest.mark.django_db
+def test_first_row(capsys):
+    AuthorRef.objects.create(name='author', id=1, cqrs_revision=0, cqrs_updated=now())
+
+    call_command(COMMAND_NAME, '--cqrs-id=author')
+
+    captured = capsys.readouterr()
+    assert '{},'.format(AuthorRef.CQRS_ID) in captured.out
+
+
+@pytest.mark.django_db
+def test_objects_less_than_batch(capsys):
+    AuthorRef.objects.create(name='author', id=1, cqrs_revision=0, cqrs_updated=now())
+
+    call_command(COMMAND_NAME, '--cqrs-id=author')
+
+    captured = capsys.readouterr()
+    out_lines = captured.out.split('\n')
+
+    assert ujson.loads(out_lines[1]) == [1]
+
+
+@pytest.mark.django_db
+def test_objects_more_than_batch(capsys):
+    for i in range(3):
+        AuthorRef.objects.create(name=str(i), id=i, cqrs_revision=i, cqrs_updated=now())
+
+    call_command(COMMAND_NAME, '--cqrs-id=author', '--batch=2')
+
+    captured = capsys.readouterr()
+    out_lines = captured.out.split('\n')
+    assert ujson.loads(out_lines[1]) == [0, 1]
+    assert ujson.loads(out_lines[2]) == [2]
+
+
+@pytest.mark.django_db
+def test_filter_no_objects(capsys):
+    call_command(COMMAND_NAME, '--cqrs-id=author', '-f={"id__in": [1, 2]}')
+
+    captured = capsys.readouterr()
+    assert 'No objects found for filter!' in captured.err
+
+
+@pytest.mark.django_db
+def test_objects_are_filtered(capsys):
+    for i in range(2):
+        AuthorRef.objects.create(name=str(i), id=i, cqrs_revision=i, cqrs_updated=now())
+
+    call_command(COMMAND_NAME, '--cqrs-id=author', '-f={"id__in": [1, 3]}')
+
+    captured = capsys.readouterr()
+    out_lines = captured.out.split('\n')
+
+    assert ujson.loads(out_lines[1]) == [1]