feat(output): Allow keepalive config (#378)

ethpandaops · Sep 20, 2024 · ab67593 · ab67593
1 parent 7417dae
commit ab67593
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 5 deletions.
diff --git a/pkg/output/xatu/config.go b/pkg/output/xatu/config.go
@@ -16,6 +16,13 @@ type Config struct {
 	Workers             int               `yaml:"workers" default:"1"`
 	Retry               RetryConfig       `yaml:"retry"`
 	AuthorizationSecret string            `yaml:"authorizationSecret"`
+	KeepAlive           KeepAliveConfig   `yaml:"keepAlive"`
+}
+
+type KeepAliveConfig struct {
+	Enabled bool          `yaml:"enabled" default:"true"`
+	Time    time.Duration `yaml:"time" default:"10s"`
+	Timeout time.Duration `yaml:"timeout" default:"30s"`
 }
 
 func (c *Config) Validate() error {

diff --git a/pkg/output/xatu/exporter.go b/pkg/output/xatu/exporter.go
@@ -32,14 +32,26 @@ type ItemExporter struct {
 }
 
 func NewItemExporter(name string, config *Config, log logrus.FieldLogger) (ItemExporter, error) {
+	log = log.WithField("output_name", name).WithField("output_type", SinkType)
+
 	opts := []grpc.DialOption{
 		grpc.WithChainUnaryInterceptor(grpc_prometheus.UnaryClientInterceptor, retry.UnaryClientInterceptor()),
 		grpc.WithChainStreamInterceptor(grpc_prometheus.StreamClientInterceptor, retry.StreamClientInterceptor()),
-		grpc.WithKeepaliveParams(keepalive.ClientParameters{
-			Time:                10 * time.Second,
-			Timeout:             30 * time.Second,
+	}
+
+	if config.KeepAlive.Enabled {
+		log.
+			WithField("keepalive_time", config.KeepAlive.Time).
+			WithField("keepalive_timeout", config.KeepAlive.Timeout).
+			Info("Enabling keepalive")
+
+		opts = append(opts, grpc.WithKeepaliveParams(keepalive.ClientParameters{
+			Time:                config.KeepAlive.Time,
+			Timeout:             config.KeepAlive.Timeout,
 			PermitWithoutStream: true,
-		}),
+		}))
+	} else {
+		log.Info("Disabling keepalive")
 	}
 
 	if config.TLS {
@@ -60,7 +72,7 @@ func NewItemExporter(name string, config *Config, log logrus.FieldLogger) (ItemE
 
 	return ItemExporter{
 		config: config,
-		log:    log.WithField("output_name", name).WithField("output_type", SinkType),
+		log:    log,
 		conn:   conn,
 		client: pb.NewEventIngesterClient(conn),
 	}, nil