Merge pull request #10037 from caseydavenport/casey-daemon-fv

Add basic daemon FV

Author: marvin-tigera

goldmane/fv/daemon_test.go

@@ -0,0 +1,272 @@
+package fv
+
+import (
+	"context"
+	"crypto/x509"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	. "github.com/onsi/gomega"
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+
+	"github.com/projectcalico/calico/goldmane/pkg/client"
+	"github.com/projectcalico/calico/goldmane/pkg/daemon"
+	"github.com/projectcalico/calico/goldmane/pkg/testutils"
+	"github.com/projectcalico/calico/goldmane/proto"
+	"github.com/projectcalico/calico/lib/std/cryptoutils"
+	"github.com/projectcalico/calico/libcalico-go/lib/logutils"
+)
+
+var (
+	ctx         context.Context
+	goldmaneURL string
+	clientCA    string
+	clientCert  string
+	clientKey   string
+
+	emitted *emissionCounter
+)
+
+func daemonSetup(t *testing.T, cfg daemon.Config) func() {
+	RegisterTestingT(t)
+	logrus.SetLevel(logrus.DebugLevel)
+	logutils.ConfigureFormatter("daemonfv")
+	logCancel := logutils.RedirectLogrusToTestingT(t)
+
+	// The context acts as a global timeout for the test to make sure we don't hang.
+	var cancel context.CancelFunc
+	ctx, cancel = context.WithTimeout(context.Background(), 30*time.Second)
+
+	// Create TLS credentials for Goldmane.
+	cert, key := createKeyCertPair(os.TempDir())
+
+	// Create TLS credentials for the client.
+	cliCert, cliKey := createKeyCertPair(os.TempDir())
+
+	// Store the file paths for the client certificates.
+	clientCA = cert.Name()
+	clientKey = cliKey.Name()
+	clientCert = cliCert.Name()
+
+	// Augment the configuration with the paths to the certificates.
+	cfg.ServerCertPath = cert.Name()
+	cfg.ServerKeyPath = key.Name()
+	cfg.CACertPath = cliCert.Name()
+
+	// Start a test HTTP server that we can point the emitter at to verify
+	// flows are being emitted.
+	emitted = &emissionCounter{}
+	testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		logrus.WithField("path", r.URL.Path).Info("[TEST] Received request")
+		emitted.Inc()
+	}))
+	cfg.PushURL = testServer.URL
+
+	// Run the daemon.
+	go daemon.Run(ctx, cfg)
+
+	goldmaneURL = fmt.Sprintf("localhost:%d", cfg.Port)
+
+	return func() {
+		logCancel()
+		cancel()
+	}
+}
+
+type emissionCounter struct {
+	sync.Mutex
+	flows int
+}
+
+func (t *emissionCounter) Inc() {
+	t.Lock()
+	defer t.Unlock()
+	t.flows++
+}
+
+func (t *emissionCounter) Count() int {
+	t.Lock()
+	defer t.Unlock()
+	return t.flows
+}
+
+func createKeyCertPair(dir string) (*os.File, *os.File) {
+	certPEM, keyPEM, err := cryptoutils.GenerateSelfSignedCert(
+		cryptoutils.WithDNSNames("localhost"),
+		cryptoutils.WithExtKeyUsages(x509.ExtKeyUsageAny),
+	)
+	Expect(err).ShouldNot(HaveOccurred())
+
+	certFile, err := os.CreateTemp(dir, "cert.pem")
+	Expect(err).ShouldNot(HaveOccurred())
+	defer certFile.Close()
+
+	keyFile, err := os.CreateTemp(dir, "key.pem")
+	Expect(err).ShouldNot(HaveOccurred())
+	defer keyFile.Close()
+
+	_, err = certFile.Write(certPEM)
+	Expect(err).ShouldNot(HaveOccurred())
+	_, err = keyFile.Write(keyPEM)
+	Expect(err).ShouldNot(HaveOccurred())
+
+	return certFile, keyFile
+}
+
+// TestDaemonCanary acts as a baseline test to ensure we can start the daemon and connect to it.
+// If this test fails, it likely means something fundamental is wrong.
+func TestDaemonCanary(t *testing.T) {
+	cfg := daemon.Config{
+		LogLevel:                 "debug",
+		Port:                     8988,
+		AggregationWindow:        time.Second * 1,
+		EmitAfterSeconds:         2,
+		EmitterAggregationWindow: time.Second * 2,
+	}
+	defer daemonSetup(t, cfg)()
+
+	// Generate credentials for the Goldmane client.
+	creds, err := client.ClientCredentials(clientCert, clientKey, clientCA)
+	if err != nil {
+		logrus.WithError(err).Fatal("Failed to create goldmane TLS credentials.")
+	}
+
+	// Verify we can connect to the server.
+	cli, err := client.NewFlowsAPIClient(goldmaneURL, grpc.WithTransportCredentials(creds))
+	require.NoError(t, err)
+	require.NotNil(t, cli)
+
+	// Verify we can list flows.
+	Eventually(func() error {
+		_, err = cli.List(ctx, nil)
+		return err
+	}, 5*time.Second, 1*time.Second).Should(Succeed())
+}
+
+// TestFlows tests that we can ingest flows, that they show up in List reqeusts, and that they
+// are emitted to the configured endpoint.
+func TestFlows(t *testing.T) {
+	cfg := daemon.Config{
+		LogLevel:                 "debug",
+		Port:                     8988,
+		AggregationWindow:        time.Second * 1,
+		EmitAfterSeconds:         2,
+		EmitterAggregationWindow: time.Second * 2,
+	}
+	defer daemonSetup(t, cfg)()
+
+	// Generate credentials for the Goldmane client.
+	creds, err := client.ClientCredentials(clientCert, clientKey, clientCA)
+	if err != nil {
+		logrus.WithError(err).Fatal("Failed to create goldmane TLS credentials.")
+	}
+
+	// Create a client to interact with Flows.
+	cli, err := client.NewFlowsAPIClient("localhost:8988", grpc.WithTransportCredentials(creds))
+	require.NoError(t, err)
+	require.NotNil(t, cli)
+
+	// Create a client to pusher Flows.
+	pusher, err := client.NewFlowClient(goldmaneURL, clientCert, clientKey, clientCA)
+	require.NoError(t, err)
+
+	connected := pusher.Connect(ctx)
+	require.NoError(t, err)
+	Eventually(connected, 5*time.Second, 100*time.Millisecond).Should(BeClosed())
+
+	// Start a goroutine to continuously send flows.
+	go func(ctx context.Context) {
+		for {
+			if ctx.Err() != nil {
+				return
+			}
+			f := testutils.NewRandomFlow(time.Now().Unix())
+			pusher.Push(f)
+			time.Sleep(100 * time.Millisecond)
+		}
+	}(ctx)
+
+	// Verify we can list flows.
+	var flows []*proto.FlowResult
+	Eventually(func() error {
+		flows, err = cli.List(ctx, nil)
+		if err != nil {
+			return err
+		}
+		if len(flows) == 0 {
+			return fmt.Errorf("no flows returned")
+		}
+		return nil
+	}, 5*time.Second, 1*time.Second).Should(Succeed())
+
+	// We should eventually see flows emitted.
+	// Sincse we only emit after 2 seconds with an emitter aggregation window of 2 seconds, we
+	// should see at least one flow emitted after 4 seconds. We'll wait for 10 seconds to be sure.
+	Eventually(emitted.Count, 10*time.Second, 1*time.Second).Should(BeNumerically(">", 0))
+}
+
+// TestHints tests that we can successfully retrieve hints from generated flows.
+func TestHints(t *testing.T) {
+	cfg := daemon.Config{
+		LogLevel:                 "debug",
+		Port:                     8988,
+		AggregationWindow:        time.Second * 1,
+		EmitAfterSeconds:         2,
+		EmitterAggregationWindow: time.Second * 2,
+	}
+	defer daemonSetup(t, cfg)()
+
+	// Generate credentials for the Goldmane client.
+	creds, err := client.ClientCredentials(clientCert, clientKey, clientCA)
+	if err != nil {
+		logrus.WithError(err).Fatal("Failed to create goldmane TLS credentials.")
+	}
+
+	// Create a client to interact with Flows.
+	cli, err := client.NewFlowsAPIClient("localhost:8988", grpc.WithTransportCredentials(creds))
+	require.NoError(t, err)
+	require.NotNil(t, cli)
+
+	// Create a client to pusher Flows.
+	pusher, err := client.NewFlowClient(goldmaneURL, clientCert, clientKey, clientCA)
+	require.NoError(t, err)
+
+	connected := pusher.Connect(ctx)
+	require.NoError(t, err)
+	Eventually(connected, 5*time.Second, 100*time.Millisecond).Should(BeClosed())
+
+	// Start a goroutine to continuously send flows.
+	go func(ctx context.Context) {
+		for {
+			if ctx.Err() != nil {
+				return
+			}
+			f := testutils.NewRandomFlow(time.Now().Unix())
+			pusher.Push(f)
+			time.Sleep(100 * time.Millisecond)
+		}
+	}(ctx)
+
+	// Verify we can list flows.
+	var hints []*proto.FilterHint
+	req := &proto.FilterHintsRequest{
+		Type: proto.FilterType_FilterTypeDestNamespace,
+	}
+	Eventually(func() error {
+		hints, err = cli.FiltersHints(ctx, req)
+		if err != nil {
+			return err
+		}
+		if len(hints) == 0 {
+			return fmt.Errorf("no hints returned")
+		}
+		return nil
+	}, 5*time.Second, 1*time.Second).Should(Succeed())
+}

goldmane/pkg/aggregator/bucketing/bucket_ring.go

@@ -329,6 +329,11 @@ func (r *BucketRing) EmitFlowCollections(sink Sink) {
 	for i := len(collections) - 1; i >= 0; i-- {
 		c := collections[i]
 		if len(c.Flows) > 0 {
+			logrus.WithFields(logrus.Fields{
+				"start": c.StartTime,
+				"end":   c.EndTime,
+				"num":   len(c.Flows),
+			}).Debug("Emitting flow collection")
 			sink.Receive(c)
 			c.Complete()
 		}

goldmane/pkg/client/tls.go

@@ -16,8 +16,10 @@ package client
 import (
 	"crypto/tls"
 	"crypto/x509"
+	"fmt"
 	"os"
 
+	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc/credentials"
 )
 
@@ -34,15 +36,20 @@ func ClientCredentials(cert, key, ca string) (credentials.TransportCredentials,
 
 func tlsConfig(cert, key, caFile string) (*tls.Config, error) {
 	// Load client cert.
+	logrus.WithFields(logrus.Fields{
+		"cert": cert,
+		"key":  key,
+	}).Debug("Loading client cert and key")
 	certificate, err := tls.LoadX509KeyPair(cert, key)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to load keypair: %s", err)
 	}
 
 	// Load CA cert.
+	logrus.WithField("ca", caFile).Debug("Loading CA cert")
 	caCert, err := os.ReadFile(caFile)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to load CA: %s", err)
 	}
 	caCertPool := x509.NewCertPool()
 	caCertPool.AppendCertsFromPEM(caCert)

goldmane/pkg/emitter/emitter.go

@@ -111,7 +111,9 @@ func (e *Emitter) Run(ctx context.Context) {
 		defer e.q.ShutDown()
 		select {
 		case <-ctx.Done():
+			logrus.Info("Context cancelled, shutting down emitter.")
 		case <-done:
+			logrus.Info("Emitter shutting down.")
 		}
 	}()
 
@@ -130,7 +132,7 @@ func (e *Emitter) Run(ctx context.Context) {
 		// Get pending work from the queue.
 		key, quit := e.q.Get()
 		if quit {
-			logrus.WithField("cm", configMapKey).Info("Emitter shutting down.")
+			logrus.Info("Emitter queue completed")
 			return
 		}
 		e.q.Done(key)