feat: add influxdb trace collection (#970)

* feat: add influxdb trace collection to the e2e test

* fix: add default toml field for batch size

* fix: nil panics when starting or stoping the client

* fix: nil clients

* fix: silly network naming issue

* chore: docs

Co-authored-by: Sanaz Taheri <35961250+staheri14@users.noreply.github.com>

* refactor: stateevent -> setevent

Co-authored-by: Rootul P <rootulp@gmail.com>

* chore: remove the default collection of each consensus step

* refactor: use "celestia" as default org

Co-authored-by: Callum Waters <cmwaters19@gmail.com>

* feat: add validate basic and IsCollecting method

* chore: actually validate basic

* chore: add influx db flags to main command

* refactor: change config to be part of the instrumentation config

* fix: forgot to update these as well

* fix: consolidate configs

* chore: remove unused const

* chore: rename remote -> trace

* chore: edit readme and copy into go doc

* chore: remove redundant logs

---------

Co-authored-by: Sanaz Taheri <35961250+staheri14@users.noreply.github.com>
Co-authored-by: Rootul P <rootulp@gmail.com>
Co-authored-by: Callum Waters <cmwaters19@gmail.com>

cmd/tendermint/commands/run_node.go

@@ -12,6 +12,7 @@ import (
 	cfg "github.com/tendermint/tendermint/config"
 	tmos "github.com/tendermint/tendermint/libs/os"
 	nm "github.com/tendermint/tendermint/node"
+	"github.com/tendermint/tendermint/pkg/trace"
 )
 
 var (
@@ -93,6 +94,19 @@ func AddNodeFlags(cmd *cobra.Command) {
 		"db_dir",
 		config.DBPath,
 		"database directory")
+
+	cmd.PersistentFlags().String(
+		trace.FlagInfluxDBURL,
+		config.Instrumentation.InfluxURL,
+		trace.FlagInfluxDBURLDescription,
+	)
+
+	cmd.PersistentFlags().String(
+		trace.FlagInfluxDBToken,
+		config.Instrumentation.InfluxToken,
+		trace.FlagInfluxDBTokenDescription,
+	)
+
 }
 
 // NewRunNodeCmd returns the command that allows the CLI to start a node.

config/config.go

@@ -1158,6 +1158,21 @@ type InstrumentationConfig struct {
 
 	// Instrumentation namespace.
 	Namespace string `mapstructure:"namespace"`
+
+	// InfluxURL is the influxdb url.
+	InfluxURL string `mapstructure:"influx_url"`
+
+	// InfluxToken is the influxdb token.
+	InfluxToken string `mapstructure:"influx_token"`
+
+	// InfluxOrg is the influxdb organization.
+	InfluxOrg string `mapstructure:"influx_org"`
+
+	// InfluxBucket is the influxdb bucket.
+	InfluxBucket string `mapstructure:"influx_bucket"`
+
+	// InfluxBatchSize is the number of points to write in a single batch.
+	InfluxBatchSize int `mapstructure:"influx_batch_size"`
 }
 
 // DefaultInstrumentationConfig returns a default configuration for metrics
@@ -1168,6 +1183,10 @@ func DefaultInstrumentationConfig() *InstrumentationConfig {
 		PrometheusListenAddr: ":26660",
 		MaxOpenConnections:   3,
 		Namespace:            "tendermint",
+		InfluxURL:            "",
+		InfluxOrg:            "celestia",
+		InfluxBucket:         "e2e",
+		InfluxBatchSize:      20,
 	}
 }
 
@@ -1183,6 +1202,23 @@ func (cfg *InstrumentationConfig) ValidateBasic() error {
 	if cfg.MaxOpenConnections < 0 {
 		return errors.New("max_open_connections can't be negative")
 	}
+	// if there is not InfluxURL configured, then we do not need to validate the rest
+	// of the config because we are not connecting.
+	if cfg.InfluxURL == "" {
+		return nil
+	}
+	if cfg.InfluxToken == "" {
+		return fmt.Errorf("token is required")
+	}
+	if cfg.InfluxOrg == "" {
+		return fmt.Errorf("org is required")
+	}
+	if cfg.InfluxBucket == "" {
+		return fmt.Errorf("bucket is required")
+	}
+	if cfg.InfluxBatchSize <= 0 {
+		return fmt.Errorf("batch size must be greater than 0")
+	}
 	return nil
 }
 

config/toml.go

@@ -535,6 +535,22 @@ max_open_connections = {{ .Instrumentation.MaxOpenConnections }}
 
 # Instrumentation namespace
 namespace = "{{ .Instrumentation.Namespace }}"
+
+# The URL of the influxdb instance to use for remote event 
+# collection. If empty, remote event collection is disabled.
+influx_url = "{{ .Instrumentation.InfluxURL }}"
+
+# The influxdb token to use for remote event collection.
+influx_token = "{{ .Instrumentation.InfluxToken }}"
+
+# The influxdb bucket to use for remote event collection.
+influx_bucket = "{{ .Instrumentation.InfluxBucket }}"
+
+# The influxdb org to use for event remote collection.
+influx_org = "{{ .Instrumentation.InfluxOrg }}"
+
+# The size of the batches that are sent to the database.
+influx_batch_size = {{ .Instrumentation.InfluxBatchSize }}
 `
 
 /****** these are for test settings ***********/

consensus/state.go

@@ -24,6 +24,7 @@ import (
 	"github.com/tendermint/tendermint/libs/service"
 	tmsync "github.com/tendermint/tendermint/libs/sync"
 	"github.com/tendermint/tendermint/p2p"
+	"github.com/tendermint/tendermint/pkg/trace"
 	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
 	sm "github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/types"
@@ -140,6 +141,8 @@ type State struct {
 
 	// for reporting metrics
 	metrics *Metrics
+
+	eventCollector *trace.Client
 }
 
 // StateOption sets an optional parameter on the State.
@@ -170,6 +173,7 @@ func NewState(
 		evpool:           evpool,
 		evsw:             tmevents.NewEventSwitch(),
 		metrics:          NopMetrics(),
+		eventCollector:   &trace.Client{},
 	}
 
 	// set function defaults (may be overwritten before calling Start)
@@ -211,6 +215,11 @@ func StateMetrics(metrics *Metrics) StateOption {
 	return func(cs *State) { cs.metrics = metrics }
 }
 
+// SetEventCollector sets the remote event collector.
+func SetEventCollector(ec *trace.Client) StateOption {
+	return func(cs *State) { cs.eventCollector = ec }
+}
+
 // String returns a string.
 func (cs *State) String() string {
 	// better not to access shared variables

go.mod

@@ -54,6 +54,8 @@ require (
 	google.golang.org/protobuf v1.28.2-0.20220831092852-f930b1dc76e8
 )
 
+require github.com/influxdata/influxdb-client-go/v2 v2.12.2
+
 require (
 	4d63.com/gochecknoglobals v0.1.0 // indirect
 	github.com/Abirdcfly/dupword v0.0.7 // indirect
@@ -90,10 +92,10 @@ require (
 	github.com/containerd/typeurl v1.0.2 // indirect
 	github.com/cosmos/go-bip39 v0.0.0-20180819234021-555e2067c45d // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
-	github.com/creachadair/taskgroup v0.3.2
 	github.com/curioswitch/go-reassign v0.2.0 // indirect
 	github.com/daixiang0/gci v0.8.1 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/deepmap/oapi-codegen v1.8.2 // indirect
 	github.com/denis-tingaikin/go-header v0.4.3 // indirect
 	github.com/dgraph-io/badger/v2 v2.2007.2 // indirect
 	github.com/dgraph-io/ristretto v0.0.3-0.20200630154024-f66de99634de // indirect
@@ -150,6 +152,7 @@ require (
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/hexops/gotextdiff v1.0.3 // indirect
 	github.com/inconshreveable/mousetrap v1.0.1 // indirect
+	github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839 // indirect
 	github.com/jdxcode/netrc v0.0.0-20210204082910-926c7f70242a // indirect
 	github.com/jgautheron/goconst v1.5.1 // indirect
 	github.com/jingyugao/rowserrcheck v1.1.1 // indirect
@@ -203,7 +206,6 @@ require (
 	github.com/polyfloyd/go-errorlint v1.0.5 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.32.1 // indirect
-	github.com/prometheus/procfs v0.8.0 // indirect
 	github.com/quasilyte/go-ruleguard v0.3.18 // indirect
 	github.com/quasilyte/gogrep v0.0.0-20220828223005-86e4605de09f // indirect
 	github.com/quasilyte/regex/syntax v0.0.0-20200407221936-30656e2c4a95 // indirect
@@ -272,3 +274,8 @@ require (
 	mvdan.cc/lint v0.0.0-20170908181259-adc824a0674b // indirect
 	mvdan.cc/unparam v0.0.0-20220706161116-678bad134442 // indirect
 )
+
+require (
+	github.com/creachadair/taskgroup v0.3.2
+	github.com/prometheus/procfs v0.8.0 // indirect
+)