116davinder / apache-kafka-backup-and-restore Goto Github PK

It will take backup of given topic and store that into either local filesystem or S3.It will also restore from given local filesystem or S3.

License: Other

Python 93.10% Mustache 6.90%

backup restore kafka apache-kafka python3 linux s3 aws-backup threading consumer

apache-kafka-backup-and-restore's Introduction

A DevOps Ninja, who can do SDLC / Big-Data / Cloud / Not AI/ML :)

apache-kafka-backup-and-restore's People

Contributors

Stargazers

Watchers

Forkers

vligade grglzrv

apache-kafka-backup-and-restore's Issues

"[Errno 2] No such file or directory: '/tmp/<topic name>/0/20221013-133145.tar.gz

Getting the following error

{ "@timestamp": "2022-10-13 11:48:40,328","level": "INFO","thread": "Kafka Consumer 1","name": "root","message": "started polling on <topic name>" }                                                                                                                                               │
│ { "@timestamp": "2022-10-13 11:48:40,328","level": "INFO","thread": "Kafka Consumer 0","name": "root","message": "started polling on <topic name>" }                                                                                                                                               │
│ { "@timestamp": "2022-10-13 11:48:40,623","level": "INFO","thread": "MainThread","name": "botocore.credentials","message": "Found credentials from IAM Role: <role name>" }                                                                                                             │
│ { "@timestamp": "2022-10-13 11:48:40,726","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 11:50:20,825","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 11:52:00,931","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 11:52:01,587","level": "INFO","thread": "S3 Upload Threads","name": "root","message": "upload successful at s3://<s3 bucket name>/<<topic name>>/0/20221013-115127.tar.gz" }                                                                                    │
│ { "@timestamp": "2022-10-13 11:52:58,904","level": "INFO","thread": "Kafka Consumer 0","name": "root","message": "Created Successful Backupfile /tmp/<topic name>/0/20221013-115127.tar.gz" }                                                                                                      │
│ { "@timestamp": "2022-10-13 11:52:58,904","level": "ERROR","thread": "Kafka Consumer 0","name": "root","message": "[Errno 2] No such file or directory: '/tmp/<topic name>/0/20221013-115127.tar.gz'" }                                                                                            │
│ { "@timestamp": "2022-10-13 11:53:41,031","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }
{ "@timestamp": "2022-10-13 12:50:24,321","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 12:52:04,422","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 12:53:44,522","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 12:55:24,614","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 12:57:04,714","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 12:58:44,813","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:00:24,913","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:02:05,005","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:03:45,105","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:05:25,205","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:07:05,304","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:08:45,404","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:10:25,501","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:12:05,601","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:13:45,701","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:15:25,801","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:17:05,902","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:18:45,961","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:20:26,061","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:22:06,161","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:23:46,261","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:25:26,358","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:27:06,457","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:28:46,557","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:30:26,652","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:32:06,753","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }                                                                                                                                                           │
│ { "@timestamp": "2022-10-13 13:32:07,312","level": "INFO","thread": "S3 Upload Threads","name": "root","message": "upload successful at s3://<s3 bucket name>/ <topic name>/0/20221013-133145.tar.gz" }                                                                                    │
│ { "@timestamp": "2022-10-13 13:32:28,669","level": "INFO","thread": "Kafka Consumer 0","name": "root","message": "Created Successful Backupfile /tmp/ <topic name>/0/20221013-133145.tar.gz" }                                                                                                      │
│ { "@timestamp": "2022-10-13 13:32:28,669","level": "ERROR","thread": "Kafka Consumer 0","name": "root","message": "[Errno 2] No such file or directory: '/tmp/ <topic name>/0/20221013-133145.tar.gz'" }                                                                                            │
│ { "@timestamp": "2022-10-13 13:33:46,825","level": "INFO","thread": "MainThread","name": "root","message": "s3 upload retry for new files in 100 seconds" }

backup.json file

{
  "BOOTSTRAP_SERVERS": "kafka01:9092,kafka02:9092,kafka03:9092",
  "TOPIC_NAMES": ["davinder.test"],
  "GROUP_ID": "Kafka-BackUp-Consumer-Group",
  "FILESYSTEM_TYPE": "S3",
  "BUCKET_NAME": "kafka-backup",
  "FILESYSTEM_BACKUP_DIR": "/tmp/",
  "NUMBER_OF_MESSAGE_PER_BACKUP_FILE": 5000000,
  "RETRY_UPLOAD_SECONDS": 100,
  "NUMBER_OF_KAFKA_THREADS": 2,
  "LOG_LEVEL": 20
}

Bug: Infinite Counter

Add reset for count: https://github.com/116davinder/apache-kafka-backup-and-restore/blob/master/backup.py#L46
Memory Issue if backup app keep running for very long time.

aws s3 restore doesn't work

davinderpal@DESKTOP-07TAJVL:~/projects/apache-kafka-backup-and-restore$ python3 restore.py example-jsons/restore-s3.json
{ "@timestamp": "2022-10-17 23:07:48,509","level": "INFO","thread": "Kafka Restore Thread","name": "root","message": "retry for more files in /tmp/davinder.test after 100" }
{ "@timestamp": "2022-10-17 23:07:48,513","level": "INFO","thread": "S3 Download","name": "botocore.credentials","message": "Found credentials in environment variables." }
Exception in thread S3 Download:
Traceback (most recent call last):
  File "/usr/lib/python3.10/threading.py", line 1009, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.10/threading.py", line 946, in run
    self._target(*self._args, **self._kwargs)
  File "/mnt/c/Users/Davinder/Documents/apache-kafka-backup-and-restore/cloud/aws.py", line 120, in s3_download
    s3_client = boto3.client('s3', endpoint_url="http://localhost:4566", use_ssl=False)
  File "/home/davinderpal/.local/lib/python3.10/site-packages/boto3/__init__.py", line 92, in client
    return _get_default_session().client(*args, **kwargs)
  File "/home/davinderpal/.local/lib/python3.10/site-packages/boto3/session.py", line 299, in client
    return self._session.create_client(
  File "/home/davinderpal/.local/lib/python3.10/site-packages/botocore/session.py", line 976, in create_client
    client = client_creator.create_client(
  File "/home/davinderpal/.local/lib/python3.10/site-packages/botocore/client.py", line 115, in create_client
    cls = self._create_client_class(service_name, service_model)
  File "/home/davinderpal/.local/lib/python3.10/site-packages/botocore/client.py", line 172, in _create_client_class
    self._event_emitter.emit(
  File "/home/davinderpal/.local/lib/python3.10/site-packages/botocore/hooks.py", line 412, in emit
    return self._emitter.emit(aliased_event_name, **kwargs)
  File "/home/davinderpal/.local/lib/python3.10/site-packages/botocore/hooks.py", line 256, in emit
    return self._emit(event_name, kwargs)
  File "/home/davinderpal/.local/lib/python3.10/site-packages/botocore/hooks.py", line 239, in _emit
    response = handler(**kwargs)
  File "/home/davinderpal/.local/lib/python3.10/site-packages/boto3/utils.py", line 65, in _handler
    module = import_module(module)
  File "/home/davinderpal/.local/lib/python3.10/site-packages/boto3/utils.py", line 56, in import_module
    __import__(name)
  File "/home/davinderpal/.local/lib/python3.10/site-packages/boto3/s3/inject.py", line 16, in <module>
    from boto3.s3.transfer import (
  File "/home/davinderpal/.local/lib/python3.10/site-packages/boto3/s3/transfer.py", line 126, in <module>
    from s3transfer.exceptions import (
  File "/home/davinderpal/.local/lib/python3.10/site-packages/s3transfer/__init__.py", line 373, in <module>
    class MultipartUploader:
  File "/home/davinderpal/.local/lib/python3.10/site-packages/s3transfer/__init__.py", line 388, in MultipartUploader
    executor_cls=concurrent.futures.ThreadPoolExecutor,
  File "/usr/lib/python3.10/concurrent/futures/__init__.py", line 49, in __getattr__
    from .thread import ThreadPoolExecutor as te
  File "/usr/lib/python3.10/concurrent/futures/thread.py", line 37, in <module>
    threading._register_atexit(_python_exit)
  File "/usr/lib/python3.10/threading.py", line 1497, in _register_atexit
    raise RuntimeError("can't register atexit after shutdown")
RuntimeError: can't register atexit after shutdown
^CException ignored in: <module 'threading' from '/usr/lib/python3.10/threading.py'>
Traceback (most recent call last):
  File "/usr/lib/python3.10/threading.py", line 1560, in _shutdown
    lock.acquire()
KeyboardInterrupt:

Add support for Alibaba Cloud Storage

https://www.alibabacloud.com/help/en/object-storage-service/latest/python-preface

Add Support for Mino Storage

https://docs.min.io/docs/python-client-quickstart-guide.html

Remove Standard Datetime Module with https://pendulum.eustace.io/

Remove all datatime common module with pendulum because of security Issues

add support for IBM CoS

https://ibm.github.io/ibm-cos-sdk-python/reference/services/s3.html
https://medium.com/@snehalgawas/working-with-ibm-cloud-object-storage-in-python-fe0ba8667d5f

Profiling + Optimization + Documentation

Ref:

https://www.youtube.com/watch?v=m_a0fN48Alw
https://jiffyclub.github.io/snakeviz/
https://www.infoworld.com/article/3600993/9-fine-libraries-for-profiling-python-code.html
Fail Fast: Move https://github.com/116davinder/apache-kafka-backup-and-restore/blob/master/cloud/minio.py#L180 to this location https://github.com/116davinder/apache-kafka-backup-and-restore/blob/master/cloud/minio.py#L145
Create separate docs folder

Reduce Code Length of Backup.py and restore.py

move duplicate code/checks to common module

Idea

Move Kbackup and Krestore Class under kafka folder

add support for blackblaze storage

Add support for digital ocean spaces

https://docs.digitalocean.com/products/spaces/reference/s3-sdk-examples/

restore script logic issue

The below Function only returns the count and later we use for on this count variable to fetch backup files.

    def s3_count_partitions(s3_client,bucket,topic):
        """It will return number of objects in a given s3 bucket and s3 bucket path."""

        try:
            return s3_client.list_objects_v2(
                Bucket=bucket,
                Prefix=topic + "/",
                Delimiter='/'
            )['KeyCount']
        except NoCredentialsError as e:
            logging.error(e)
            exit(1)

Edge Case:
Initial Partition Count: 10
Backup script copies partitions : 0,1,4,6,7,8 only
Restore script aka above-mentioned function will return Count: 6

Now when following for loop on for p in range(_pc): will only consider 0,1,2,3,4,5 partitions and few of them won't even
exists in S3 so it will keep failing for them

    def s3_download(bucket,topic,tmp_dir,retry_download_seconds=60):
        s3_client = boto3.client('s3')
        while True:
            _pc = Download.s3_count_partitions(s3_client,bucket,topic)
            # create temp. topic directory
            for p in range(_pc):
                os.makedirs(os.path.join(tmp_dir,topic,str(p)),exist_ok=True)

            for p in range(_pc):
                os.makedirs(os.path.join(tmp_dir,topic,str(p)),exist_ok=True)

            for _pt in range(_pc):

                _ck = checkpoint.read_checkpoint_partition(tmp_dir,topic,str(_pt))
                _partition_path = os.path.join(topic,str(_pt))
                _s3_partition_files = Download.s3_list_files(s3_client,bucket,_partition_path)

Error

davinderpal@DESKTOP-07TAJVL:~/projects/apache-kafka-backup-and-restore$ python3 restore.py example-jsons/restore-s3.json
{ "@timestamp": "2022-10-17 23:30:34,750","level": "INFO","thread": "Kafka Restore Thread","name": "root","message": "retry for more files in /tmp/davinder.test after 100" }
{ "@timestamp": "2022-10-17 23:30:34,853","level": "INFO","thread": "MainThread","name": "root","message": "Test messeage" }
{ "@timestamp": "2022-10-17 23:30:34,861","level": "INFO","thread": "MainThread","name": "botocore.credentials","message": "Found credentials in environment variables." }
{ "@timestamp": "2022-10-17 23:30:34,909","level": "WARNING","thread": "MainThread","name": "root","message": "[Errno 2] No such file or directory: '/tmp/davinder.test/0/checkpoint'" }
{ "@timestamp": "2022-10-17 23:30:34,931","level": "WARNING","thread": "MainThread","name": "root","message": "[Errno 2] No such file or directory: '/tmp/davinder.test/2/checkpoint'" }
{ "@timestamp": "2022-10-17 23:30:34,938","level": "WARNING","thread": "MainThread","name": "root","message": "[Errno 2] No such file or directory: '/tmp/davinder.test/3/checkpoint'" }
{ "@timestamp": "2022-10-17 23:30:34,946","level": "WARNING","thread": "MainThread","name": "root","message": "[Errno 2] No such file or directory: '/tmp/davinder.test/4/checkpoint'" }
{ "@timestamp": "2022-10-17 23:30:34,985","level": "INFO","thread": "MainThread","name": "root","message": "download success for /tmp/davinder.test/4/20221017-230228.tar.gz and its sha256 file " }
{ "@timestamp": "2022-10-17 23:30:34,985","level": "WARNING","thread": "MainThread","name": "root","message": "[Errno 2] No such file or directory: '/tmp/davinder.test/5/checkpoint'" }
{ "@timestamp": "2022-10-17 23:30:34,993","level": "INFO","thread": "MainThread","name": "root","message": "retry for new file after 100s in s3://kafka-backup/davinder.test" }

Potential Solution:
Instead of returning the count of partitions, we can return the actual number of partitons but with regex or split method to extract from
list_objects_v2 method call