Docker demo

docker/Dockerfile

@@ -0,0 +1,20 @@
+FROM golang:stretch as build
+
+RUN go get github.com/Masterminds/glide && \
+    go get github.com/alecthomas/gometalinter && \
+    go get github.com/tinylib/msgp
+
+RUN apt-get update && apt-get install -y netcat
+
+RUN mkdir -p $GOPATH/src/github.com/uber && \
+    cd $GOPATH/src/github.com/uber && \
+    git clone https://github.com/uber/storagetapper.git && \
+    cd storagetapper && \
+    rm -f pipe/hdfs.go && \
+    DEB_BUILD_OPTIONS=nocheck make install
+
+ADD wait_for.sh /wait_for.sh
+RUN chmod +x /wait_for.sh
+
+VOLUME ['/etc/storagetapper/']
+ENTRYPOINT ['/wait_for.sh', 'storagetapper']

docker/Dockerfile_demo

@@ -0,0 +1,10 @@
+FROM debian:stretch
+
+RUN apt-get update && apt-get install -y netcat mysql-client curl
+
+ADD wait_for.sh /wait_for.sh
+ADD demo.sh /demo.sh
+RUN chmod +x /wait_for.sh /demo.sh
+
+ENTRYPOINT ["/wait_for.sh", "/demo.sh"]
+CMD [""]

docker/Dockerfile_kafkaconsumer

@@ -0,0 +1,7 @@
+FROM wurstmeister/kafka
+
+ADD wait_for.sh /wait_for.sh
+RUN chmod +x /wait_for.sh
+
+ENTRYPOINT ["/wait_for.sh", "/opt/kafka/bin/kafka-console-consumer.sh"]
+CMD ['']

docker/README.md

@@ -0,0 +1,4 @@
+Docker Demo
+-----------
+
+A docker-compose highlighting how to run storagetapper and capture snapshot/changes for an mysql table

docker/config/config-file.cnf

@@ -0,0 +1,8 @@
+[mysqld]
+server-id=1
+binlog-format=ROW
+gtid_mode=ON
+enforce-gtid-consistency
+log_slave_updates=1
+log-bin=bin.log
+log-bin-index=bin-log.index

docker/config/production.yaml

@@ -0,0 +1,8 @@
+log_type: std
+state_update_timeout: 10
+
+state_connect_url: root:storagetapper@db
+kafka_addresses:
+  - "kafka:9092"
+changelog_topic_name_template_default: "storagetapper.changelog.{{.Service}}.{{.Db}}.{{.Table}}"
+output_topic_name_template_default: "storagetapper.output.{{.Service}}.{{.Db}}.{{.Table}}"

docker/demo.sh

@@ -0,0 +1,48 @@
+#!/bin/bash
+
+set -e
+
+sql() {
+    db=$1
+    shift
+    mysql -uroot -pstoragetapper -hdb "$db" -e "$@"
+}
+
+
+sql "" "DROP DATABASE IF EXISTS ex_db1"
+sql "" "RESET MASTER"
+
+sql "" "CREATE DATABASE ex_db1"
+sql "ex_db1" "CREATE TABLE ex_table1(id int not null primary key, ts TIMESTAMP, type varchar(32))"
+
+for i in `seq 1 10`; do
+    sql "ex_db1" "INSERT INTO ex_table1(id, type) VALUES ($i, 'pre-existing')"
+done
+
+
+curl -s --data '{"cmd" : "add", "name" : "ex_cluster1", "host" : "db", "port" : 3306, "user" : "root", "pw" : "storagetapper"}' http://storagetapper:7836/cluster
+curl -s --data '{"cmd" : "add", "cluster" : "ex_cluster1", "service" : "ex_svc1", "db":"ex_db1", "table":"ex_table1", "output":"kafka", "outputFormat":"json"}' http://storagetapper:7836/table
+
+
+sleep 12
+
+for j in `seq 1 100`; do
+
+let start=$j*100
+let end=$start+10
+
+for i in `seq $start $end`; do
+    sql "ex_db1" "INSERT INTO ex_table1(id, type) VALUES ($i, 'incremental')"
+    sleep 1
+done
+
+sql "ex_db1" "ALTER TABLE ex_table1 ADD schema_change varchar(32)"
+
+for i in `seq $start $end`; do
+    sql "ex_db1" "UPDATE ex_table1 SET type = 'update' WHERE id = $i;"
+    sleep 1
+done
+
+sql "ex_db1" "ALTER TABLE ex_table1 DROP schema_change"
+
+done

docker/docker-compose.yml

@@ -0,0 +1,61 @@
+version: '3.1'
+
+services:
+
+  db:
+    image: mysql:5
+    command: --default-authentication-plugin=mysql_native_password
+    restart: always
+    environment:
+      # Use root/storagetapper as user/password credentials
+      MYSQL_ROOT_PASSWORD: storagetapper
+    volumes:
+      - ./config:/etc/mysql/conf.d/
+    security_opt: 
+      - seccomp:unconfined 
+
+  adminer:
+    image: adminer
+    restart: always
+    ports:
+      - 8080:8080
+
+  zookeeper:
+    image: wurstmeister/zookeeper
+
+  kafka:
+    image: wurstmeister/kafka
+    environment:
+      KAFKA_CREATE_TOPICS: "storagetapper.changelog.ex_svc1.ex_db1.ex_table1:1:1,storagetapper.output.ex_svc1.ex_db1.ex_table1:1:1"
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_ADVERTISED_HOST_NAME: kafka
+
+  kafka_client_changelog:
+    build:
+        context: .
+        dockerfile: Dockerfile_kafkaconsumer
+    environment:
+      - WAIT_FOR=kafka:9092
+    command: --bootstrap-server kafka:9092 --topic storagetapper.changelog.ex_svc1.ex_db1.ex_table1 --from-beginning
+
+  kafka_client_output:
+    build:
+        context: .
+        dockerfile: Dockerfile_kafkaconsumer
+    environment:
+      - WAIT_FOR=kafka:9092
+    command: --bootstrap-server kafka:9092 --topic storagetapper.output.ex_svc1.ex_db1.ex_table1 --from-beginning    
+
+  storagetapper:
+    build: .
+    environment:
+      - WAIT_FOR=db:3306;kafka:9092
+    volumes:
+      - ./config:/etc/storagetapper/
+
+  demo:
+    build:
+        context: .
+        dockerfile: Dockerfile_demo
+    environment:
+      - WAIT_FOR=storagetapper:7836

docker/wait_for.sh

@@ -0,0 +1,15 @@
+#!/bin/bash
+
+if [ -n "$WAIT_FOR" ]; then
+	IFS=';' read -a HOSTPORT_ARRAY <<< "$WAIT_FOR"
+	for HOSTPORT in "${HOSTPORT_ARRAY[@]}"
+	do
+		WAIT_FOR_HOST=${HOSTPORT%:*}
+		WAIT_FOR_PORT=${HOSTPORT#*:}
+
+		echo Waiting for $WAIT_FOR_HOST to listen on $WAIT_FOR_PORT...
+		while ! nc -z $WAIT_FOR_HOST $WAIT_FOR_PORT; do echo sleeping; sleep 2; done
+	done
+fi
+
+exec "$@"

scripts/workflow_example.sh

@@ -35,7 +35,7 @@ trap "kill $TPID; exit 1" SIGINT SIGTERM SIGHUP
 sleep 2
 
 curl --data '{"cmd" : "add", "name" : "ex_cluster1", "host" : "localhost", "port" : 3306, "user" : "root", "pw" : ""}' http://localhost:7836/cluster
-curl --data '{"cmd" : "add", "cluster" : "ex_cluster1", "service" : "ex_svc1", "db":"ex_db1", "table":"ex_table1"}' http://localhost:7836/table
+curl --data '{"cmd" : "add", "cluster" : "ex_cluster1", "service" : "ex_svc1", "db":"ex_db1", "table":"ex_table1", "output":"kafka", "outputFormat":"json"}' http://localhost:7836/table
 
 sleep 12