Merge pull request #41 from ethpandaops/feat/tls

feat: tls

example_discovery.yaml

@@ -3,6 +3,7 @@ metricsAddr: ":9090"
 
 coordinator:
   address: localhost:8080
+  tls: false
   headers:
     Authorization: Someb64Value
   max_queue_size: 51200

example_mimicry.yaml

@@ -31,7 +31,7 @@ coordinator:
   #   max_peers: 100
 
 outputs:
-- name: basic
+- name: http-sink
   type: http
   config:
     address: http://localhost:8080
@@ -41,3 +41,14 @@ outputs:
     batch_timeout: 5s
     export_timeout: 30s
     max_export_batch_size: 512
+- name: xatu-server
+  type: xatu
+  config:
+    address: localhost:8080
+    tls: false
+    headers:
+      Authorization: Someb64Value
+    max_queue_size: 51200
+    batch_timeout: 5s
+    export_timeout: 30s
+    max_export_batch_size: 512

pkg/discovery/coordinator/config.go

@@ -8,6 +8,7 @@ import (
 type Config struct {
 	Address                  string            `yaml:"address"`
 	Headers                  map[string]string `yaml:"headers"`
+	TLS                      bool              `yaml:"tls" default:"false"`
 	MaxQueueSize             int               `yaml:"max_queue_size" default:"51200"`
 	BatchTimeout             time.Duration     `yaml:"batch_timeout" default:"5s"`
 	ExportTimeout            time.Duration     `yaml:"export_timeout" default:"30s"`

pkg/discovery/coordinator/exporter.go

@@ -3,10 +3,12 @@ package coordinator
 import (
 	"context"
 	"fmt"
+	"net"
 
 	pb "github.com/ethpandaops/xatu/pkg/proto/xatu"
 	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/encoding/gzip"
 	"google.golang.org/grpc/metadata"
@@ -22,7 +24,17 @@ type ItemExporter struct {
 
 func NewItemExporter(config *Config, log logrus.FieldLogger) (ItemExporter, error) {
 	var opts []grpc.DialOption
-	opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+
+	if config.TLS {
+		host, _, err := net.SplitHostPort(config.Address)
+		if err != nil {
+			return ItemExporter{}, fmt.Errorf("fail to get host from address: %v", err)
+		}
+
+		opts = append(opts, grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, host)))
+	} else {
+		opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	}
 
 	conn, err := grpc.Dial(config.Address, opts...)
 	if err != nil {

pkg/output/xatu/config.go

@@ -8,6 +8,7 @@ import (
 type Config struct {
 	Address            string            `yaml:"address"`
 	Headers            map[string]string `yaml:"headers"`
+	TLS                bool              `yaml:"tls" default:"false"`
 	MaxQueueSize       int               `yaml:"max_queue_size" default:"51200"`
 	BatchTimeout       time.Duration     `yaml:"batch_timeout" default:"5s"`
 	ExportTimeout      time.Duration     `yaml:"export_timeout" default:"30s"`

pkg/output/xatu/exporter.go

@@ -3,10 +3,12 @@ package xatu
 import (
 	"context"
 	"fmt"
+	"net"
 
 	pb "github.com/ethpandaops/xatu/pkg/proto/xatu"
 	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/encoding/gzip"
 	"google.golang.org/grpc/metadata"
@@ -22,7 +24,17 @@ type ItemExporter struct {
 
 func NewItemExporter(config *Config, log logrus.FieldLogger) (ItemExporter, error) {
 	var opts []grpc.DialOption
-	opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+
+	if config.TLS {
+		host, _, err := net.SplitHostPort(config.Address)
+		if err != nil {
+			return ItemExporter{}, fmt.Errorf("fail to get host from address: %v", err)
+		}
+
+		opts = append(opts, grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, host)))
+	} else {
+		opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	}
 
 	conn, err := grpc.Dial(config.Address, opts...)
 	if err != nil {