Added health check endpoint (#18)

Author: pparshin

Makefile

@@ -16,7 +16,7 @@ lint:
 
 .PHONY: run
 run: build
-	bin/${BINARY} -config=config/replicator.conf.yml
+	bin/${BINARY} -config=config/dev.conf.yml
 
 .PHONY: run_short_tests
 run_short_tests:

README.md

@@ -32,3 +32,13 @@ a list of replicated columns, a space name.
 Replicator reads primary keys from MySQL table info and sync them automatically.
 Updating primary key in MySQL causes two Tarantool requests: delete an old row and insert a new one, because
 it is illegal to update primary key in Tarantool.
+
+## Metrics
+
+Replicator exposes several debug endpoints:
+
+* `/metrics` - runtime and app metrics in Prometheus format,
+* `/health` - health check.
+
+Health check returns status `503 Service Unavailable` if replicator is not running, dumping 
+data or replication lag greater than `app.health.seconds_behind_master` config value.

cmd/replicator/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"flag"
 	"fmt"
 	"io"
@@ -13,6 +14,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/etherlabsio/healthcheck"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
@@ -48,7 +50,13 @@ func main() {
 
 	metrics.Init()
 
-	server := initHTTPServer(cfg.App.ListenAddr)
+	b, err := bridge.New(cfg, logger)
+	if err != nil {
+		logger.Fatal().Err(err).Msg("could not establish bridge from MySQL to Tarantool")
+	}
+
+	health := initHealthHandler(cfg.App.Health, b)
+	server := initHTTPServer(cfg.App.ListenAddr, health)
 	go func() {
 		logger.Info().Msgf("listening on %s", cfg.App.ListenAddr)
 
@@ -58,14 +66,9 @@ func main() {
 		}
 	}()
 
-	b, err := bridge.New(cfg, logger)
-	if err != nil {
-		logger.Fatal().Err(err).Msg("could not establish bridge from MySQL to Tarantool")
-	}
-
 	go func() {
-		errors := b.Run()
-		for errRun := range errors {
+		runErrors := b.Run()
+		for errRun := range runErrors {
 			logger.Err(errRun).Msg("got sync error")
 		}
 
@@ -155,14 +158,52 @@ func newRollingLogFile(cfg *config.Logging) (io.Writer, error) {
 	}, nil
 }
 
-func initHTTPServer(addr string) *http.Server {
+func initHTTPServer(addr string, health http.Handler) *http.Server {
 	server := &http.Server{
 		Addr:         addr,
 		ReadTimeout:  5 * time.Second, //nolint:gomnd
 		WriteTimeout: 5 * time.Second, //nolint:gomnd
 	}
 
 	http.Handle("/metrics", promhttp.Handler())
+	http.Handle("/health", health)
 
 	return server
 }
+
+func initHealthHandler(cfg config.Health, b *bridge.Bridge) http.Handler {
+	sbm := uint32(cfg.SecondsBehindMaster)
+
+	return healthcheck.Handler(
+		healthcheck.WithChecker(
+			"lag", healthcheck.CheckerFunc(
+				func(ctx context.Context) error {
+					cur := b.Delay()
+					if cur > sbm {
+						return fmt.Errorf("replication lag too big: %d", cur)
+					}
+
+					return nil
+				},
+			),
+		),
+
+		healthcheck.WithChecker(
+			"state", healthcheck.CheckerFunc(
+				func(ctx context.Context) error {
+					dumping := b.Dumping()
+					if dumping {
+						return errors.New("replicator has not yet finished dump process")
+					}
+
+					running := b.Running()
+					if !running {
+						return errors.New("replication is not running")
+					}
+
+					return nil
+				},
+			),
+		),
+	)
+}

config/dev.conf.yml

@@ -0,0 +1,47 @@
+app:
+  listen_addr: ':8080'
+  data_file: '/tmp/mysql-tarantool/state.info'
+  health:
+    seconds_behind_master: 10
+  logging:
+    level: 'debug'
+    syslog_enabled: false
+    file_enabled: false
+    file_name: '/var/log/mysql-tarantool-replicator.log'
+    file_max_size: 256
+    file_max_backups: 3
+    file_max_age: 5
+
+replication:
+  server_id: 100
+  gtid_mode: true
+
+  mysql:
+    dump:
+      exec_path: '/usr/bin/mysqldump'
+      skip_master_data: false
+      extra_options:
+        - '--column-statistics=0'
+    addr: '127.0.0.1:13306'
+    user: 'repl'
+    password: 'repl'
+    charset: 'utf8'
+
+  tarantool:
+    addr: '127.0.0.1:13301'
+    user: 'repl'
+    password: 'repl'
+    max_retries: 3
+    connect_timeout: '500ms'
+    request_timeout: '500ms'
+
+  mappings:
+    - source:
+        schema: 'city'
+        table: 'users'
+        columns:
+          - username
+          - password
+          - email
+      dest:
+        space: 'users'

config/replicator.conf.yml

@@ -1,6 +1,8 @@
 app:
   listen_addr: ':8080'
   data_file: '/etc/mysql-tarantool/state.info'
+  health:
+    seconds_behind_master: 10
   logging:
     level: 'debug'
     syslog_enabled: false

go.mod

@@ -4,6 +4,7 @@ go 1.14
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/etherlabsio/healthcheck v0.0.0-20191224061800-dd3d2fd8c3f6
 	github.com/philhofer/fwd v1.1.0 // indirect
 	github.com/prometheus/client_golang v1.8.0
 	github.com/rs/zerolog v1.20.0
@@ -14,6 +15,7 @@ require (
 	github.com/stretchr/testify v1.6.1
 	github.com/tinylib/msgp v1.1.2 // indirect
 	github.com/viciious/go-tarantool v0.0.0-20201014090959-d4e1044f393b
+	go.uber.org/atomic v1.5.0
 	golang.org/x/sys v0.0.0-20201029080932-201ba4db2418
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
 	gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c

go.sum

@@ -63,6 +63,8 @@ github.com/edsrzf/mmap-go v1.0.0/go.mod h1:YO35OhQPt3KJa3ryjFM5Bs14WD66h8eGKpfaB
 github.com/envoyproxy/go-control-plane v0.6.9/go.mod h1:SBwIajubJHhxtWwsL9s8ss4safvEdbitLhGGK48rN6g=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
+github.com/etherlabsio/healthcheck v0.0.0-20191224061800-dd3d2fd8c3f6 h1:az9jaEKre+mwUWiS9Pl8h1FuOvdiFM7UqplmCmJtHUQ=
+github.com/etherlabsio/healthcheck v0.0.0-20191224061800-dd3d2fd8c3f6/go.mod h1:ZMSmptAGNIg5UAxsJzmw5DMW6uQvxr/hvCklNwtFz1k=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/franela/goblin v0.0.0-20200105215937-c9ffbefa60db/go.mod h1:7dvUGVsVBjqR7JHJk0brhHOZYGmfBYOrK0ZhYMEtBr4=
 github.com/franela/goreq v0.0.0-20171204163338-bcd34c9993f8/go.mod h1:ZhphrRTfi2rbfLwlschooIH4+wKKDR4Pdxhh+TRoA20=

internal/bridge/replicator.go

@@ -11,6 +11,7 @@ import (
 	"github.com/siddontang/go-mysql/canal"
 	"github.com/siddontang/go-mysql/mysql"
 	tnt "github.com/viciious/go-tarantool"
+	"go.uber.org/atomic"
 
 	"github.com/pparshin/go-mysql-tarantool/internal/config"
 	"github.com/pparshin/go-mysql-tarantool/internal/metrics"
@@ -32,13 +33,18 @@ type Bridge struct {
 	cancel context.CancelFunc
 	logger zerolog.Logger
 
+	dumping *atomic.Bool
+	running *atomic.Bool
+
 	syncCh    chan interface{}
 	closeOnce *sync.Once
 }
 
 func New(cfg *config.Config, logger zerolog.Logger) (*Bridge, error) {
 	b := &Bridge{
 		logger:    logger,
+		dumping:   atomic.NewBool(false),
+		running:   atomic.NewBool(false),
 		syncCh:    make(chan interface{}, eventsBufSize),
 		closeOnce: &sync.Once{},
 	}
@@ -238,12 +244,12 @@ func (b *Bridge) newTarantoolClient(cfg *config.Config) {
 //
 // Returns closed channel with all errors or an empty channel.
 func (b *Bridge) Run() <-chan error {
-	metrics.SetReplicationState(true)
-	defer metrics.SetReplicationState(false)
+	b.setRunning(true)
+	defer b.setRunning(false)
 
 	go func() {
 		for range time.Tick(1 * time.Second) {
-			metrics.SetSecondsBehindMaster(b.canal.GetDelay())
+			metrics.SetSecondsBehindMaster(b.Delay())
 		}
 	}()
 
@@ -266,6 +272,12 @@ func (b *Bridge) Run() <-chan error {
 		}
 	}()
 
+	b.setDumping(true)
+	go func() {
+		<-b.canal.WaitDumpDone()
+		b.setDumping(false)
+	}()
+
 	var err error
 	pos := b.stateSaver.position()
 	switch p := pos.(type) {
@@ -346,3 +358,25 @@ func (b *Bridge) Close() error {
 
 	return err
 }
+
+func (b *Bridge) Delay() uint32 {
+	return b.canal.GetDelay()
+}
+
+func (b *Bridge) setRunning(v bool) {
+	b.running.Store(v)
+	b.setDumping(false)
+	metrics.SetReplicationState(v)
+}
+
+func (b *Bridge) Running() bool {
+	return b.running.Load()
+}
+
+func (b *Bridge) setDumping(v bool) {
+	b.dumping.Store(v)
+}
+
+func (b *Bridge) Dumping() bool {
+	return b.dumping.Load()
+}

internal/bridge/testdata/cfg.yml

@@ -1,5 +1,8 @@
 app:
+  listen_addr: ':8080'
   data_file: '/tmp/mysql-tarantool-replicator/state.info'
+  health:
+    seconds_behind_master: 5
   logging:
     level: 'debug'
     syslog_enabled: false

internal/config/cfg.go

@@ -11,6 +11,7 @@ import (
 const (
 	defaultListenAddr         = ":8080"
 	defaultDataFile           = "/etc/mysql-tarantool-replicator/state.info"
+	defaultHealthSBM          = 10
 	defaultLogLevel           = "debug"
 	defaultSysLogEnabled      = false
 	defaultFileLoggingEnabled = false
@@ -46,9 +47,14 @@ type Config struct {
 type AppConfig struct {
 	ListenAddr string  `yaml:"listen_addr"`
 	DataFile   string  `yaml:"data_file"`
+	Health     Health  `yaml:"health"`
 	Logging    Logging `yaml:"logging"`
 }
 
+type Health struct {
+	SecondsBehindMaster int `yaml:"seconds_behind_master"`
+}
+
 type Logging struct {
 	Level              string `yaml:"level"`
 	SysLogEnabled      bool   `yaml:"syslog_enabled"`
@@ -65,9 +71,10 @@ func (c *AppConfig) withDefaults() {
 	}
 
 	c.ListenAddr = defaultListenAddr
-
 	c.DataFile = defaultDataFile
 
+	c.Health.SecondsBehindMaster = defaultHealthSBM
+
 	c.Logging.Level = defaultLogLevel
 	c.Logging.SysLogEnabled = defaultSysLogEnabled
 	c.Logging.FileLoggingEnabled = defaultFileLoggingEnabled