Fix issues with telemetry

cmd/cli/cli.go

@@ -108,7 +108,7 @@ func RunCli(supportedSourceConfigPairs []sourceconfig.ConfigPair) {
 		}
 		o.Start()
 
-		telemetry.InitTelemetryWithCollector(cfg)
+		stopTelemetry := telemetry.InitTelemetryWithCollector(cfg)
 
 		// Handle SIGTERM
 		sig := make(chan os.Signal)
@@ -126,7 +126,8 @@ func RunCli(supportedSourceConfigPairs []sourceconfig.ConfigPair) {
 			select {
 			case <-stop:
 				log.Debug("source.Stop() finished successfully!")
-
+
+				stopTelemetry()
 				err := common.DeleteTemporaryDir()
 				if err != nil {
 					log.Debugf(`error deleting tmp directory: %v`, err)
@@ -137,6 +138,7 @@ func RunCli(supportedSourceConfigPairs []sourceconfig.ConfigPair) {
 				t.Close()
 				ft.Close()
 				o.Stop()
+				stopTelemetry()
 
 				err := common.DeleteTemporaryDir()
 				if err != nil {

config/config.go

@@ -44,7 +44,7 @@ type configurationData struct {
 	LogLevel                string         `hcl:"log_level,optional" env:"LOG_LEVEL"`
 	GoogleServiceAccountB64 string         `hcl:"google_application_credentials_b64,optional" env:"GOOGLE_APPLICATION_CREDENTIALS_B64"`
 	UserProvidedID          string         `hcl:"user_provided_id,optional" env:"USER_PROVIDED_ID"`
-	EnableTelemetry         bool           `hcl:"enable_telemetry,optional" env:"ENABLE_TELEMETRY"`
+	DisableTelemetry        bool           `hcl:"disable_telemetry,optional" env:"DISABLE_TELEMETRY"`
 }
 
 // component is a type to abstract over configuration blocks.
@@ -98,9 +98,9 @@ func defaultConfigData() *configurationData {
 			TimeoutSec: 1,
 			BufferSec:  15,
 		},
-		Transformations: nil,
-		LogLevel:        "info",
-		EnableTelemetry: true,
+		Transformations:  nil,
+		LogLevel:         "info",
+		DisableTelemetry: false,
 	}
 }
 

pkg/telemetry/const.go

@@ -2,6 +2,8 @@ package telemetry
 
 import (
 	"time"
+
+	"github.com/snowplow-devops/stream-replicator/cmd"
 )
 
 var (
@@ -11,5 +13,5 @@ var (
 	url                = "telemetry-g.snowplowanalytics.com"
 	port               = "443"
 	applicationName    = "stream-replicator"
-	applicationVersion = "1.0.0"
+	applicationVersion = cmd.AppVersion
 )

pkg/telemetry/telemetry.go

@@ -13,7 +13,7 @@ import (
 
 // config holds the configuration for telemetry
 type config struct {
-	enable             bool
+	disable            bool
 	interval           time.Duration
 	method             string
 	url                string
@@ -27,7 +27,7 @@ type config struct {
 
 func newTelemetryWithConfig(cfg *conf.Config) *config {
 	return &config{
-		enable:             cfg.Data.EnableTelemetry,
+		disable:            cfg.Data.DisableTelemetry,
 		interval:           interval,
 		method:             method,
 		protocol:           protocol,
@@ -40,7 +40,7 @@ func newTelemetryWithConfig(cfg *conf.Config) *config {
 	}
 }
 
-func initTelemetry(telemetry *config) {
+func initTelemetry(telemetry *config) func() {
 	storage := gt.InitStorageMemory()
 	emitter := gt.InitEmitter(
 		gt.RequireCollectorUri(fmt.Sprintf(`%s:%s`, telemetry.url, telemetry.port)),
@@ -76,13 +76,24 @@ func initTelemetry(telemetry *config) {
 
 	ticker := time.NewTicker(telemetry.interval)
 
+	stop := make(chan struct{})
+
 	go func() {
 		makeAndTrackHeartbeat(telemetry, tracker)
 		for {
-			<-ticker.C
-			makeAndTrackHeartbeat(telemetry, tracker)
+			select {
+			case <-ticker.C:
+				makeAndTrackHeartbeat(telemetry, tracker)
+			case <-stop:
+				return
+			}
+
 		}
 	}()
+
+	return func() {
+		close(stop)
+	}
 }
 
 func makeAndTrackHeartbeat(telemetry *config, tracker *gt.Tracker) {
@@ -99,9 +110,12 @@ func makeAndTrackHeartbeat(telemetry *config, tracker *gt.Tracker) {
 }
 
 // InitTelemetryWithCollector initialises telemetry
-func InitTelemetryWithCollector(cfg *conf.Config) {
+func InitTelemetryWithCollector(cfg *conf.Config) func() {
 	telemetry := newTelemetryWithConfig(cfg)
-	initTelemetry(telemetry)
+	if telemetry.disable {
+		return func() {}
+	}
+	return initTelemetry(telemetry)
 }
 
 func makeHeartbeatEvent(service config) *gt.SelfDescribingJson {