Kafka Producer + Consumer

cmd/dispatch/ingest.go

@@ -3,9 +3,12 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"fmt"
+	"github.com/Shopify/sarama"
 	"github.com/kevinmichaelchen/api-dispatch/internal/idl/coop/drivers/dispatch/v1beta1"
 	"github.com/kevinmichaelchen/api-dispatch/internal/service/money"
 	"github.com/spf13/cobra"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
 	"io/ioutil"
 	"log"
@@ -70,6 +73,11 @@ func ingestDrivers(cmd *cobra.Command, args []string) {
 		})
 	}
 
+	err = sendLocations(locations)
+	if err != nil {
+		log.Fatalf("Failed to send messages to Kafka: %v", err)
+	}
+
 	log.Printf("Seeding %d driver locations\n", len(locations))
 	req := &v1beta1.UpdateDriverLocationsRequest{Locations: locations}
 	s, err := marshalProto(req)
@@ -95,6 +103,50 @@ func ingestDrivers(cmd *cobra.Command, args []string) {
 	log.Println(s)
 }
 
+func sendLocations(locations []*v1beta1.DriverLocation) error {
+	version, err := sarama.ParseKafkaVersion("3.1.1")
+	if err != nil {
+		return err
+	}
+	config := sarama.NewConfig()
+	config.Version = version
+	p, err := sarama.NewSyncProducer([]string{"127.0.0.1:9092"}, config)
+	if err != nil {
+		return err
+	}
+	msgs, err := locationsToMessages(locations)
+	if err != nil {
+		return err
+	}
+	return p.SendMessages(msgs)
+}
+
+func locationsToMessages(locations []*v1beta1.DriverLocation) ([]*sarama.ProducerMessage, error) {
+	var out []*sarama.ProducerMessage
+	for _, e := range locations {
+		m, err := locationToMessage(e)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, m)
+	}
+	return out, nil
+}
+
+func locationToMessage(location *v1beta1.DriverLocation) (*sarama.ProducerMessage, error) {
+	b, err := proto.Marshal(location)
+	if err != nil {
+		return nil, err
+	}
+
+	key := fmt.Sprintf("%s-%s", location.GetDriverId(), location.GetMostRecentHeartbeat().String())
+	return &sarama.ProducerMessage{
+		Topic: "driver-locations",
+		Key:   sarama.StringEncoder(key),
+		Value: sarama.ByteEncoder(b),
+	}, nil
+}
+
 func ingestTrips(cmd *cobra.Command, args []string) {
 	// Open file
 	f, err := os.Open(ingestPath)

docker-compose.yml

@@ -41,6 +41,38 @@ services:
       timeout: 1s
       retries: 5
 
+  zookeeper:
+    image: confluentinc/cp-zookeeper:7.1.0
+    hostname: zookeeper
+    container_name: zookeeper
+    ports:
+      - "2181:2181"
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+
+  broker:
+    image: confluentinc/cp-kafka:7.1.0
+    hostname: broker
+    container_name: broker
+    depends_on:
+      - zookeeper
+    ports:
+      - "29092:29092"
+      - "9092:9092"
+      - "9101:9101"
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_JMX_PORT: 9101
+      KAFKA_JMX_HOSTNAME: localhost
+
 volumes:
   pg_data:
   prometheus_data:

docs/ingestion.md

@@ -17,4 +17,9 @@ A trip takes the form:
 of time-series data, so maybe we can assume that somewhere downstream these
 locations are persisted to something like
 [RedisTimeSeries](https://redis.io/docs/stack/timeseries/) and periodically
-compacted into Postgres using a RedisGears Python function.
+compacted into Postgres using a RedisGears Python function.
+
+Because of their high velocity, driver locations should be ingested with Kafka.
+We use the [Shopify/sarama](https://github.com/Shopify/sarama) since 
+[kafka-go](https://github.com/segmentio/kafka-go)
+was too low-level.

go.mod

@@ -3,6 +3,7 @@ module github.com/kevinmichaelchen/api-dispatch
 go 1.18
 
 require (
+	github.com/Shopify/sarama v1.34.0
 	github.com/XSAM/otelsql v0.14.1
 	github.com/codingsince1985/geo-golang v1.8.1
 	github.com/envoyproxy/protoc-gen-validate v0.1.0
@@ -12,7 +13,7 @@ require (
 	github.com/google/go-cmp v0.5.8
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
 	github.com/kat-co/vala v0.0.0-20170210184112-42e1d8b61f12
-	github.com/kr/pretty v0.2.0
+	github.com/kr/pretty v0.3.0
 	github.com/lib/pq v1.10.5
 	github.com/paulmach/go.geo v0.0.0-20180829195134-22b514266d33
 	github.com/rs/xid v1.4.0
@@ -47,27 +48,43 @@ require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/eapache/go-resiliency v1.2.0 // indirect
+	github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 // indirect
+	github.com/eapache/queue v1.1.0 // indirect
 	github.com/ericlagergren/decimal v0.0.0-20181231230500-73749d4874d5 // indirect
 	github.com/felixge/httpsnoop v1.0.2 // indirect
 	github.com/fsnotify/fsnotify v1.5.1 // indirect
 	github.com/go-logr/logr v1.2.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/gofrs/uuid v3.2.0+incompatible // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/uuid v1.3.0 // indirect
+	github.com/hashicorp/errwrap v1.0.0 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
+	github.com/hashicorp/go-uuid v1.0.2 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/inconshreveable/mousetrap v1.0.0 // indirect
-	github.com/kr/text v0.1.0 // indirect
+	github.com/jcmturner/aescts/v2 v2.0.0 // indirect
+	github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect
+	github.com/jcmturner/gofork v1.0.0 // indirect
+	github.com/jcmturner/gokrb5/v8 v8.4.2 // indirect
+	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
+	github.com/klauspost/compress v1.15.0 // indirect
+	github.com/kr/text v0.2.0 // indirect
 	github.com/magiconair/properties v1.8.5 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
 	github.com/mitchellh/mapstructure v1.4.2 // indirect
 	github.com/paulmach/go.geojson v1.4.0 // indirect
 	github.com/pelletier/go-toml v1.9.4 // indirect
+	github.com/pierrec/lz4/v4 v4.1.14 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_golang v1.12.1 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.32.1 // indirect
 	github.com/prometheus/procfs v0.7.3 // indirect
+	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
+	github.com/rogpeppe/go-internal v1.6.1 // indirect
 	github.com/spf13/afero v1.6.0 // indirect
 	github.com/spf13/cast v1.4.1 // indirect
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect
@@ -78,7 +95,8 @@ require (
 	go.uber.org/atomic v1.9.0 // indirect
 	go.uber.org/dig v1.14.1 // indirect
 	go.uber.org/multierr v1.8.0 // indirect
-	golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4 // indirect
+	golang.org/x/crypto v0.0.0-20220214200702-86341886e292 // indirect
+	golang.org/x/net v0.0.0-20220520000938-2e3eb7b945c2 // indirect
 	golang.org/x/sys v0.0.0-20220429233432-b5fbb4746d32 // indirect
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/time v0.0.0-20200416051211-89c76fbcd5d1 // indirect