Merge pull request #24 from tinybirdco/fix_log

* avoid panic when trying to parse response
* fix/reorder logs to get better visibility of flow
* improve Dockerfile to launch in container with a mysql db

Author: asuncepeda

Dockerfile

@@ -1,13 +1,24 @@
 #FROM python:3.8
-FROM ubuntu:20.04 as build
-COPY requirements.txt requirements.txt
-RUN apt-get update && apt-get install -y python3-pip libmysqlclient-dev openssh-client mysql-client nano -y iputils-ping
+FROM ubuntu:22.04 as build
+#COPY requirements.txt requirements.txt
+
+ARG DEBIAN_FRONTEND=noninteractive
+run apt update && apt install -y mysql-server software-properties-common
+run add-apt-repository ppa:deadsnakes/ppa 
+
+RUN apt update && apt install -y python3-pip python3.8 python3.8-venv python3.8-dev libssl-dev libcurl4-openssl-dev libpython3-dev build-essential libmysqlclient-dev autossh mysql-client pkg-config
+
+run mkdir -p /mnt/disks/tb/tinybird_mysql_connector/syncer_files
+run chmod 0755 /mnt/disks/tb/tinybird_mysql_connector/
+run chmod 0755 /mnt/disks/tb/tinybird_mysql_connector/syncer_files
+
 
-RUN pip3 install --upgrade pip
-RUN pip3 install mysqlclient
-RUN pip3 install -r requirements.txt
 ENV LC_ALL=C.UTF-8
 ENV LANG=C.UTF-8
-#COPY . .
-#RUN chmod +x dev_run_config_file.sh
-#CMD ["/bin/bash", "dev_run_config_file.sh"]
+COPY . .
+
+run python3.8 -m pip wheel --wheel-dir=/tmp/clickkhouse-mysql-data-reader/ .
+run find /tmp/clickkhouse-mysql-data-reader/ -name *.whl -exec pip install {} \;
+
+ENTRYPOINT ["./init-docker.sh"]
+cmd ["--config-file=./clickhouse-mysql-docker.conf"]

clickhouse-mysql-docker.conf

@@ -0,0 +1,28 @@
+tb_host=https://api.tinybird.co
+tb_token=
+
+log_file=/mnt/disks/tb/tinybird_mysql_connector/main.log
+log_level=debug
+nice_pause=1
+
+binlog_position_file=/mnt/disks/tb/tinybird_mysql_connector/syncer_files/bl-raw_landing-local-pos
+
+mempool=yes
+mempool_max_events_num=10
+mempool_max_flush_interval=5
+csvpool=yes
+csvpool_file_path_prefix=/mnt/disks/tb/tinybird_mysql_connector/syncer_files/raw_landing_local_
+#csvpool_keep_files=yes
+pump_data=yes
+
+src_server_id=1
+src_host=127.0.0.1
+#src_port=3306
+src_user=clickhouse_mysql_reader
+src_password=1234
+src_schemas=clickhouse_mysql_reader
+src_tables=one,two
+src_wait=yes
+src_resume=yes
+
+dst_table=raw_landing

clickhouse_mysql/writer/csvwriter.py

@@ -203,6 +203,9 @@ def close(self):
     def destroy(self):
         if self.delete and os.path.isfile(self.path):
             self.close()
+            if self.next_writer_builder and self.next_writer_builder.get().not_uploaded:
+                logging.error(f"CSV { self.path } not uploaded into TB")
+                return
             os.remove(self.path)
 
 if __name__ == '__main__':

clickhouse_mysql/writer/tbcsvwriter.py

@@ -22,6 +22,8 @@ class TBCSVWriter(Writer):
     tb_host = None
     tb_token = None
 
+    not_uploaded = None
+
     def __init__(
             self,
             tb_host,
@@ -30,6 +32,7 @@ def __init__(
             dst_table=None,
             dst_table_prefix=None,
             dst_distribute=False,
+            not_uploaded=False,
     ):
         # if dst_distribute and dst_schema is not None:
         #     dst_schema += "_all"
@@ -44,16 +47,18 @@ def __init__(
         if self.tb_host is None or self.tb_token is None:
             logging.critical(
                 f" Host: {self.tb_host} or token {self.tb_token} is missing")
-            return None
+            return
 
         self.dst_schema = dst_schema
         self.dst_table = dst_table
         self.dst_table_prefix = dst_table_prefix
         self.dst_distribute = dst_distribute
+        self.not_uploaded = not_uploaded
 
 
     def uploadCSV(self, table, filename, tries=1):
         limit_of_retries = 3
+        self.not_uploaded = False
         params = {
             'name': table,
             'mode': 'append',
@@ -76,32 +81,32 @@ def uploadCSV(self, table, filename, tries=1):
                     params=params,
                     verify=False)
 
-                # logging.debug(response.text)
-                logging.info(response.json())
+                logging.info(response.content)  # this is ugly, but we need to check what is in the response for some detected errors
                 if response.status_code == 200:
                     json_object = json.loads(response.content)
                     logging.debug(f"Import id: {json_object['import_id']}")
                 elif response.status_code == 429:
                     retry_after = int(response.headers['Retry-After']) + tries
-                    logging.error(
-                        f"Too many requests retrying in {retry_after} seconds to upload {filename } to {table}")
+                    logging.error(f"Too many requests retrying in {retry_after} seconds to upload {filename} to {table}")
                     time.sleep(retry_after)
                     self.uploadCSV(table, filename, tries + 1)
                 else:
-                    # In case of error let's retry only
-                    logging.exception(response.json())
-                    time.sleep(tries)
-                    logging.info(f"Retrying { tries } of { limit_of_retries }")
+                    # In case of error let's retry only `limit_of_retries` times
+                    logging.exception(response.content)
                     if tries > limit_of_retries:
+                        self.not_uploaded = True
                         return
+                    logging.info(f"Retrying {filename} when status {response.status_code}, try {tries} of {limit_of_retries}")
+                    time.sleep(tries)
                     self.uploadCSV(table, filename, tries + 1)
         except Exception as e:
             logging.exception(e)
-            # We wait tries^2 sec to try again
-            time.sleep(tries * tries)
-            logging.info(f"Retrying { tries } of { limit_of_retries }")
             if tries > limit_of_retries:
+                self.not_uploaded = True
                 return
+            # We wait tries^2 sec to try again
+            logging.info(f"Retrying {filename} when exception: try {tries} of {limit_of_retries}")
+            time.sleep(tries * tries)
             self.uploadCSV(table, filename, tries + 1)
 
     def insert(self, event_or_events=None):
@@ -124,7 +129,7 @@ def insert(self, event_or_events=None):
         logging.debug('class:%s insert %d rows', __class__, len(events))
 
         for event in events:
-            #schema = self.dst_schema if self.dst_schema else event.schema
+            # schema = self.dst_schema if self.dst_schema else event.schema
             table = self.dst_table if self.dst_table else event.table
             self.uploadCSV(table, event.filename)
 
@@ -231,7 +236,7 @@ def update(self, event_or_events=None):
         #     )
 
         #     choptions = ""
-        #     if self.host:
+        #     if self.host::wq
         #         choptions += " --host=" + shlex.quote(self.host)
         #     if self.port:
         #         choptions += " --port=" + str(self.port)

init-docker.sh

@@ -0,0 +1,13 @@
+#!/bin/bash
+
+mysqld &
+sleep 5
+mysql -e "create user 'clickhouse_mysql_reader'@'%' identified by '1234';"
+mysql -e "create database clickhouse_mysql_reader"
+mysql -e "grant super on *.* to 'clickhouse_mysql_reader';"
+mysql -e "grant replication slave on *.* to 'clickhouse_mysql_reader';"
+
+mysql -e "use clickhouse_mysql_reader; create table one (id int, field varchar(10));"
+mysql -e "use clickhouse_mysql_reader; create table two (id int, field varchar(10));"
+
+clickhouse-mysql $@ #--config-file=./clickhouse-mysql.conf