WIP

mcascallares · Oct 1, 2019 · 498fd92 · 498fd92
1 parent 3430cb8
commit 498fd92
Show file tree

Hide file tree

Showing 20 changed files with 2,145 additions and 148 deletions.
diff --git a/mqtt-tracker/copy_from_ccloud.sh b/mqtt-tracker/copy_from_ccloud.sh
@@ -2,12 +2,19 @@
 
 source .env
 
+while [ 1 -eq 1 ] 
+do
 kafkacat -b $CCLOUD_BROKER_HOST \
     -X security.protocol=SASL_SSL -X sasl.mechanisms=PLAIN \
     -X sasl.username="$CCLOUD_API_KEY" -X sasl.password="$CCLOUD_API_SECRET" \
     -X ssl.ca.location=/usr/local/etc/openssl/cert.pem -X api.version.request=true \
-    -X auto.offset.reset=earliest
-    -G copy_to_local_00 data_mqtt -K: | \
+    -X auto.offset.reset=earliest \
+    -K: \
+    -G asgard03_copy_to_local_05 data_mqtt | \
 kafkacat -b localhost:9092,localhost:19092,localhost:29092 \
-    -t data_mqtt_ccloud_kc_00 \
-    -K: -P 
+    -t data_mqtt-import \
+    -K: -P -T
+
+sleep 30
+
+done
diff --git a/mqtt-tracker/create_es_sink.sh b/mqtt-tracker/create_es_sink.sh
@@ -2,24 +2,6 @@
 
 source .env
 
-curl -i -X PUT -H  "Content-Type:application/json" \
-      http://localhost:8083/connectors/sink-elastic-runner_status-00/config \
-      -d '{
-            "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
-            "connection.url": "'$ELASTIC_URL'",
-            "connection.username": "'$ELASTIC_USERNAME'",
-            "connection.password": "'$ELASTIC_PASSWORD'",
-            "type.name": "type.name=kafkaconnect",
-            "behavior.on.malformed.documents": "warn",
-            "topics": "RUNNER_STATUS",
-            "key.ignore": "false",
-            "schema.ignore": "true",
-            "key.converter": "org.apache.kafka.connect.storage.StringConverter",
-            "transforms": "addTS",
-            "transforms.addTS.type": "org.apache.kafka.connect.transforms.InsertField$Value",
-            "transforms.addTS.timestamp.field": "EVENT_TS"
-            }'
-
 curl -i -X PUT -H  "Content-Type:application/json" \
       http://localhost:8083/connectors/sink-elastic-runner_location-00/config \
       -d '{
@@ -29,11 +11,12 @@ curl -i -X PUT -H  "Content-Type:application/json" \
             "connection.password": "'$ELASTIC_PASSWORD'",
             "type.name": "type.name=kafkaconnect",
             "behavior.on.malformed.documents": "warn",
-            "topics": "RUNNER_LOCATION",
+            "errors.tolerance": "all",
+            "errors.log.enable":true,
+            "errors.log.include.messages":true,
+            "topics.regex": "RUNNER_LOCATION_.*",
             "key.ignore": "true",
             "schema.ignore": "true",
-            "key.converter": "org.apache.kafka.connect.storage.StringConverter",
-            "transforms": "addTS",
-            "transforms.addTS.type": "org.apache.kafka.connect.transforms.InsertField$Value",
-            "transforms.addTS.timestamp.field": "EVENT_TS"
-            }'
+            "key.converter": "org.apache.kafka.connect.storage.StringConverter"
+            }'
+
diff --git a/mqtt-tracker/create_kibana_metadata.sh b/mqtt-tracker/create_kibana_metadata.sh
@@ -0,0 +1,13 @@
+#!/bin/bash
+
+echo -e "\n--\n+> Create Kibana index patterns"
+curl -XPOST 'http://localhost:5601/api/saved_objects/index-pattern/runner_location' \
+    -H 'kbn-xsrf: nevergonnagiveyouup' \
+    -H 'Content-Type: application/json' \
+    -d '{"attributes":{"title":"runner_location*","timeFieldName":"EVENT_TIME_EPOCH_MS_TS"}}'
+
+echo -e "\n--\n+> Set default Kibana index"
+curl -XPOST 'http://localhost:5601/api/kibana/settings' \
+    -H 'kbn-xsrf: nevergonnagiveyouup' \
+    -H 'content-type: application/json' \
+    -d '{"changes":{"defaultIndex":"runner_location"}}'
diff --git a/mqtt-tracker/create_replicator_source.sh b/mqtt-tracker/create_replicator_source.sh
@@ -14,7 +14,7 @@ done
 #
 epoch=$(date +%s)
 curl -s -X PUT -H "Accept:application/json" \
-    -H  "Content-Type:application/json" http://localhost:58083/connectors/replicator-source/config \
+    -H  "Content-Type:application/json" "http://localhost:58083/connectors/replicator-source"$epoch"/config" \
     -d '
         {
         "connector.class": "io.confluent.connect.replicator.ReplicatorSourceConnector",
@@ -28,9 +28,11 @@ curl -s -X PUT -H "Accept:application/json" \
         "src.consumer.group.id": "replicator-'$epoch'",
         "dest.kafka.bootstrap.servers": "kafka-1:39092,kafka-2:49092,kafka-3:59092",
         "topic.whitelist": "data_mqtt",
-        "topic.rename.format":"${topic}-ccloud-'$epoch'",
+        "topic.rename.format":"${topic}-import",
         "confluent.license":"",
         "confluent.topic.bootstrap.servers":"kafka-1:39092,kafka-2:49092,kafka-3:59092",
         "confluent.topic.replication.factor":1,
         "offset.start":"consumer"
         }' | jq '.'
+
+#         "topic.rename.format":"${topic}-ccloud-'$epoch'",
diff --git a/mqtt-tracker/data/ddl/users.sql b/mqtt-tracker/data/ddl/users.sql
@@ -0,0 +1,26 @@
+GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'replicator' IDENTIFIED BY 'replpass';
+GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT  ON *.* TO 'debezium' IDENTIFIED BY 'dbz';
+
+CREATE DATABASE demo;
+GRANT ALL PRIVILEGES ON demo.* TO 'mysqluser'@'%';
+
+use demo;
+
+CREATE TABLE USERS (USERID VARCHAR(255), 
+                    EMAIL  VARCHAR(255), 
+                    SHARE_LOCATION_OPTIN BOOLEAN,
+                    PRIVACY_LOCATION_LAT DOUBLE,
+                    PRIVACY_LOCATION_LON DOUBLE,
+                    PRIVACY_ZONE_KM INT,
+                    CREATE_TS TIMESTAMP DEFAULT CURRENT_TIMESTAMP ,
+                    UPDATE_TS TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
+);
+
+INSERT INTO USERS (USERID, EMAIL, SHARE_LOCATION_OPTIN, PRIVACY_LOCATION_LAT, PRIVACY_LOCATION_LON, PRIVACY_ZONE_KM) 
+           VALUES ('rmoff','[email protected]', TRUE, 52.924651779196246,-1.7996157809808713, 1);
+
+INSERT INTO USERS (USERID, EMAIL, SHARE_LOCATION_OPTIN) 
+           VALUES ('tom','[email protected]', FALSE);
+
+INSERT INTO USERS (USERID, EMAIL, SHARE_LOCATION_OPTIN) 
+           VALUES ('isla','[email protected]', FALSE);
diff --git a/mqtt-tracker/data/dummy_data.kcat b/mqtt-tracker/data/dummy_data.kcat
@@ -0,0 +1,15 @@
+owntracks/race-write/ivor:{"batt":100,"lon":-1.8125752571133549,"acc":65,"p":98.489105224609375,"bs":3,"vac":10,"lat":53.955233261289684,"t":"t","conn":"w","tst":1569334836,"alt":98,"_type":"location","tid":"C1"}
+owntracks/race-write/hugh:{"cog":193,"batt":45,"lon":-78.74988541880019,"acc":16,"p":100.14521789550781,"bs":1,"vel":0,"vac":3,"lat":35.66231724270073,"conn":"w","tst":1569330852,"tid":"RM","_type":"location","alt":104}
+owntracks/race-write/rick:{"batt":100,"lon":-1.8125821847113142,"acc":65,"bs":2,"p":98.428581237792969,"vac":10,"lat":53.955247763099669,"conn":"w","tst":1569334160,"alt":98,"_type":"location","tid":"A4"}
+owntracks/tiqmyral/rmoff:{"batt":97,"lon":-1.8125818890032339,"acc":200,"p":98.689468383789062,"bs":1,"vel":0,"vac":93,"lat":53.92535400390625,"t":"u","conn":"w","tst":1569316069,"alt":97,"_type":"location","tid":"FF"}
+owntracks/race-write/hugh:{"cog":24,"batt":96,"lon":-1.5314248613277124,"acc":10,"p":99.762825012207031,"bs":2,"vel":130,"vac":4,"lat":53.645720385215718,"conn":"m","tst":1569689674,"tid":"RM","_type":"location","alt":52}
+owntracks/race-write/ivor:{"cog":19,"batt":96,"lon":-1.5301563222094541,"acc":10,"p":99.829292297363281,"bs":2,"vel":134,"vac":6,"lat":53.647593543379635,"conn":"m","tst":1569689680,"tid":"RM","_type":"location","alt":51}
+owntracks/race-write/rick:{"cog":15,"batt":96,"lon":-1.5292111820733143,"acc":10,"p":99.908317565917969,"bs":2,"vel":129,"vac":4,"lat":53.64943998483686,"conn":"m","tst":1569689686,"tid":"RM","_type":"location","alt":42}
+owntracks/race-write/ivor:{"cog":10,"batt":96,"lon":-1.5284748555519652,"acc":10,"p":99.969696044921875,"bs":2,"vel":131,"vac":4,"lat":53.651334385427887,"conn":"m","tst":1569689692,"tid":"RM","_type":"location","alt":39}
+owntracks/race-write/hugh:{"cog":3,"batt":96,"lon":-1.5280530303414841,"acc":10,"p":100.00237274169922,"bs":2,"vel":125,"vac":6,"lat":53.653228086413485,"conn":"m","tst":1569689698,"tid":"RM","_type":"location","alt":37}
+owntracks/race-write/ivor:{"cog":349,"batt":96,"lon":-1.5283093337051832,"acc":5,"p":100.01512145996094,"bs":2,"vel":125,"vac":3,"lat":53.655085952807283,"conn":"m","tst":1569689704,"tid":"RM","_type":"location","alt":36}
+owntracks/race-write/ivor:{"cog":336,"batt":96,"lon":-1.5292491484563615,"acc":5,"p":100.01145172119141,"bs":2,"vel":125,"vac":3,"lat":53.656871188023594,"conn":"m","tst":1569689710,"tid":"RM","_type":"location","alt":39}
+owntracks/race-write/rmoff:{"cog":327,"batt":96,"lon":-1.5307541122276733,"acc":5,"p":100.00465393066406,"bs":2,"vel":120,"vac":3,"lat":53.658466731924534,"conn":"m","tst":1569689716,"tid":"RM","_type":"location","alt":42}
+owntracks/race-write/rmoff:{"cog":326,"batt":96,"lon":-1.5326987256024216,"acc":5,"p":99.988372802734375,"bs":2,"vel":119,"vac":4,"lat":53.660191805074049,"conn":"m","tst":1569689723,"tid":"RM","_type":"location","alt":40}
+owntracks/race-write/rmoff:{"cog":326,"batt":96,"lon":-1.5346830820408368,"acc":5,"p":99.980300903320312,"bs":2,"vel":119,"vac":6,"lat":53.661916155530776,"conn":"m","tst":1569689730,"tid":"RM","_type":"location","alt":36}
+owntracks/race-write/rmoff:{"cog":326,"batt":96,"lon":-1.536664103040333,"acc":5,"p":99.97674560546875,"bs":2,"vel":118,"vac":6,"lat":53.663641107896694,"conn":"m","tst":1569689737,"tid":"RM","_type":"location","alt":35}