Add e2e tests

.github/workflows/go.yml

@@ -29,4 +29,4 @@ jobs:
       run: go build -mod=vendor -v ./...
 
     - name: Test
-      run: go test -race -mod=vendor -v ./...
+      run: go test -race -mod=vendor -v -timeout=30m ./...

Makefile

@@ -40,7 +40,7 @@ clean:
 .PHONY: clean
 
 test:
-	go test ./...
+	go test -timeout 30m ./...
 .PHONY: test
 
 default:

pkg/crd/crd_test.go

@@ -14,10 +14,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/testcontainers/testcontainers-go"
 	"github.com/testcontainers/testcontainers-go/modules/k3s"
-	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
-	"k8s.io/client-go/tools/clientcmd"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 type TestStore struct {
@@ -45,10 +42,6 @@ func (s *TestStore) Count() int32 {
 }
 
 func TestCRD(t *testing.T) {
-	scheme := clientgoscheme.Scheme
-	require.NoError(t, clientgoscheme.AddToScheme(scheme))
-	require.NoError(t, v1.AddToScheme(scheme))
-
 	crdPath := filepath.Join(t.TempDir(), "crd.yaml")
 	require.NoError(t, testutils.CopyFile("../../kustomize/bases/functions_crd.yaml", crdPath))
 	fnCrdPath := filepath.Join(t.TempDir(), "fn-crd.yaml")
@@ -62,16 +55,7 @@ func TestCRD(t *testing.T) {
 	require.NoError(t, k3sContainer.CopyFileToContainer(ctx, crdPath, filepath.Join(testutils.K3sManifests, "crd.yaml"), 0644))
 	require.NoError(t, k3sContainer.CopyFileToContainer(ctx, fnCrdPath, filepath.Join(testutils.K3sManifests, "fn-crd.yaml"), 0644))
 
-	kubeconfig, err := testutils.WriteKubeConfig(ctx, k3sContainer, t.TempDir())
-	require.NoError(t, err)
-
-	config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
-	require.NoError(t, err)
-	ctrlCli, err := ctrlclient.New(config, ctrlclient.Options{
-		WarningHandler: ctrlclient.WarningHandlerOptions{
-			SuppressWarnings: true,
-		},
-	})
+	_, ctrlCli, err := testutils.GetKubeConfig(ctx, k3sContainer)
 	require.NoError(t, err)
 
 	ts := &TestStore{t: t}

pkg/testutils/collector/collector.go

@@ -2,7 +2,6 @@ package collector
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"path/filepath"
 
@@ -101,12 +100,20 @@ func WithCluster(ctx context.Context, k *k3s.K3sContainer) testcontainers.Custom
 	}
 }
 
-// TODO Accept a collector configuration and write it to a temp directory, then transfer it to the cluster,
-// overwriting the existing configuration. Also create a default configuration if none is provided.
-// Note, this would require moving the configuration to somewhere it can be imported, get feedback
-// from the team.
-func WithConfig(ctx context.Context) testcontainers.CustomizeRequestOption {
-	return func(req *testcontainers.GenericContainerRequest) error {
-		return errors.New("not implemented")
+type KustoTableSchema struct{}
+
+func (k *KustoTableSchema) TableName() string {
+	return "Collector"
+}
+
+func (k *KustoTableSchema) CslColumns() []string {
+	return []string{
+		"msg:string",
+		"lvl:string",
+		"ts:datetime",
+		"namespace:string",
+		"container:string",
+		"pod:string",
+		"host:string",
 	}
 }

pkg/testutils/ingestor/ingestor.go

@@ -99,3 +99,21 @@ func WithCluster(ctx context.Context, k *k3s.K3sContainer) testcontainers.Custom
 		return nil
 	}
 }
+
+type KustoTableSchema struct{}
+
+func (k *KustoTableSchema) TableName() string {
+	return "Collector"
+}
+
+func (k *KustoTableSchema) CslColumns() []string {
+	return []string{
+		"msg:string",
+		"lvl:string",
+		"ts:datetime",
+		"namespace:string",
+		"container:string",
+		"pod:string",
+		"host:string",
+	}
+}

pkg/testutils/ingestor/k8s.yaml

@@ -115,7 +115,6 @@ spec:
           command:
             - /ingestor
           args:
-            - "--kustainer"
             - "--storage-dir=/mnt/data"
             - "--max-segment-age=5s"
             - "--max-disk-usage=21474836480"

pkg/testutils/integration_test.go

@@ -0,0 +1,106 @@
+package testutils_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/Azure/adx-mon/pkg/testutils"
+	"github.com/Azure/adx-mon/pkg/testutils/collector"
+	"github.com/Azure/adx-mon/pkg/testutils/ingestor"
+	"github.com/Azure/adx-mon/pkg/testutils/kustainer"
+	"github.com/stretchr/testify/require"
+	"github.com/testcontainers/testcontainers-go"
+	"github.com/testcontainers/testcontainers-go/modules/k3s"
+)
+
+func TestIntegration(t *testing.T) {
+	// An extra generous timeout for the test. The test should run in
+	// about 5 minutes, but when running with the race detector, it
+	// can take longer.
+	ctx, cancel := context.WithTimeout(context.Background(), time.Hour)
+	defer cancel()
+
+	k3sContainer, err := k3s.Run(ctx, "rancher/k3s:v1.31.2-k3s1")
+	testcontainers.CleanupContainer(t, k3sContainer)
+	require.NoError(t, err)
+
+	kustoContainer, err := kustainer.Run(ctx, "mcr.microsoft.com/azuredataexplorer/kustainer-linux:latest", kustainer.WithCluster(ctx, k3sContainer))
+	testcontainers.CleanupContainer(t, kustoContainer)
+	require.NoError(t, err)
+
+	restConfig, _, err := testutils.GetKubeConfig(ctx, k3sContainer)
+	require.NoError(t, err)
+	require.NoError(t, kustoContainer.PortForward(ctx, restConfig))
+
+	kubeconfig, err := testutils.WriteKubeConfig(ctx, k3sContainer, t.TempDir())
+	require.NoError(t, err)
+	t.Logf("Kubeconfig: %s", kubeconfig)
+	t.Logf("Kustainer: %s", kustoContainer.ConnectionUrl())
+
+	t.Run("Create databases", func(t *testing.T) {
+		for _, dbName := range []string{"Metrics", "Logs"} {
+			require.NoError(t, kustoContainer.CreateDatabase(ctx, dbName))
+		}
+	})
+
+	ingestorContainer, err := ingestor.Run(ctx, ingestor.WithCluster(ctx, k3sContainer))
+	testcontainers.CleanupContainer(t, ingestorContainer)
+	require.NoError(t, err)
+
+	collectorContainer, err := collector.Run(ctx, collector.WithCluster(ctx, k3sContainer))
+	testcontainers.CleanupContainer(t, collectorContainer)
+	require.NoError(t, err)
+
+	t.Run("Logs", func(t *testing.T) {
+		var (
+			pollInterval = time.Second
+			timeout      = 5 * time.Minute
+			database     = "Logs"
+			table        = "Collector"
+		)
+
+		t.Run("Table exists in Kusto", func(t *testing.T) {
+			require.Eventually(t, func() bool {
+				return testutils.TableExists(ctx, t, database, table, kustoContainer.ConnectionUrl())
+			}, timeout, pollInterval)
+		})
+
+		t.Run("Table has rows", func(t *testing.T) {
+			require.Eventually(t, func() bool {
+				return testutils.TableHasRows(ctx, t, database, table, kustoContainer.ConnectionUrl())
+			}, timeout, pollInterval)
+		})
+
+		t.Run("View exists in Kusto", func(t *testing.T) {
+			require.Eventually(t, func() bool {
+				return testutils.FunctionExists(ctx, t, database, table, kustoContainer.ConnectionUrl())
+			}, timeout, pollInterval)
+		})
+
+		t.Run("Verify view schema", func(t *testing.T) {
+			testutils.VerifyTableSchema(ctx, t, database, table, kustoContainer.ConnectionUrl(), &collector.KustoTableSchema{})
+		})
+	})
+
+	t.Run("Metrics", func(t *testing.T) {
+		var (
+			pollInterval = time.Second
+			timeout      = 5 * time.Minute
+			database     = "Metrics"
+			table        = "AdxmonCollectorHealthCheck"
+		)
+
+		t.Run("Table exists in Kusto", func(t *testing.T) {
+			require.Eventually(t, func() bool {
+				return testutils.TableExists(ctx, t, database, table, kustoContainer.ConnectionUrl())
+			}, timeout, pollInterval)
+		})
+
+		t.Run("Table has rows", func(t *testing.T) {
+			require.Eventually(t, func() bool {
+				return testutils.TableHasRows(ctx, t, database, table, kustoContainer.ConnectionUrl())
+			}, timeout, pollInterval)
+		})
+	})
+}

pkg/testutils/k8s.go

@@ -6,7 +6,12 @@ import (
 	"os"
 	"path/filepath"
 
+	v1 "github.com/Azure/adx-mon/api/v1"
 	"github.com/testcontainers/testcontainers-go/modules/k3s"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 const K3sManifests = "/var/lib/rancher/k3s/server/manifests/"
@@ -22,3 +27,31 @@ func WriteKubeConfig(ctx context.Context, k *k3s.K3sContainer, dir string) (stri
 	}
 	return kubeConfigPath, nil
 }
+
+func GetKubeConfig(ctx context.Context, k *k3s.K3sContainer) (*rest.Config, ctrlclient.Client, error) {
+	scheme := clientgoscheme.Scheme
+	if err := clientgoscheme.AddToScheme(scheme); err != nil {
+		return nil, nil, fmt.Errorf("failed to add client-go scheme: %w", err)
+	}
+	if err := v1.AddToScheme(scheme); err != nil {
+		return nil, nil, fmt.Errorf("failed to add v1 scheme: %w", err)
+	}
+
+	kubeConfigYaml, err := k.GetKubeConfig(ctx)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to get kubeconfig: %w", err)
+	}
+
+	restCfg, err := clientcmd.RESTConfigFromKubeConfig(kubeConfigYaml)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to get rest config: %w", err)
+	}
+
+	ctrlCli, err := ctrlclient.New(restCfg, ctrlclient.Options{
+		WarningHandler: ctrlclient.WarningHandlerOptions{
+			SuppressWarnings: true,
+		},
+	})
+
+	return restCfg, ctrlCli, nil
+}

pkg/testutils/kql_verify.go

@@ -2,6 +2,7 @@ package testutils
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"testing"
 
@@ -10,6 +11,11 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+type TableSchema interface {
+	TableName() string
+	CslColumns() []string
+}
+
 func TableExists(ctx context.Context, t *testing.T, database, table, uri string) bool {
 	t.Helper()
 
@@ -59,6 +65,13 @@ type Table struct {
 func FunctionExists(ctx context.Context, t *testing.T, database, function, uri string) bool {
 	t.Helper()
 
+	fn := GetFunction(ctx, t, database, function, uri)
+	return fn.Name == function
+}
+
+func GetFunction(ctx context.Context, t *testing.T, database, function, uri string) Function {
+	t.Helper()
+
 	cb := kusto.NewConnectionStringBuilder(uri)
 	client, err := kusto.New(cb)
 	require.NoError(t, err)
@@ -88,11 +101,11 @@ func FunctionExists(ctx context.Context, t *testing.T, database, function, uri s
 			continue
 		}
 		if fn.Name == function {
-			return true
+			return fn
 		}
 	}
 
-	return false
+	return Function{}
 }
 
 type Function struct {
@@ -137,10 +150,63 @@ func TableHasRows(ctx context.Context, t *testing.T, database, table, uri string
 		return count.Count > 0
 	}
 
-	t.Logf("Table %s has no rows", table)
 	return false
 }
 
 type RowCount struct {
 	Count int64 `kusto:"Count"`
 }
+
+func VerifyTableSchema(ctx context.Context, t *testing.T, database, table, uri string, expect TableSchema) {
+	t.Helper()
+
+	cb := kusto.NewConnectionStringBuilder(uri)
+	client, err := kusto.New(cb)
+	require.NoError(t, err)
+	defer client.Close()
+
+	query := kql.New("").AddUnsafe(table).AddLiteral(" | getschema")
+	rows, err := client.Query(ctx, database, query)
+	require.NoError(t, err)
+	defer rows.Stop()
+
+	var schema []*KqlSchema
+	for {
+		row, errInline, errFinal := rows.NextRowOrError()
+		if errFinal == io.EOF {
+			break
+		}
+		if errInline != nil {
+			t.Logf("Partial failure to retrieve schema: %v", errInline)
+			continue
+		}
+		if errFinal != nil {
+			t.Errorf("Failed to retrieve schema: %v", errFinal)
+		}
+
+		var s KqlSchema
+		if err := row.ToStruct(&s); err != nil {
+			t.Errorf("Failed to convert row to struct: %v", err)
+			continue
+		}
+		schema = append(schema, &s)
+	}
+
+	require.Equal(t, expect.TableName(), table)
+	require.Equal(t, cslSchemaFromKqlSchema(schema), expect.CslColumns())
+}
+
+type KqlSchema struct {
+	ColumnName    string `kusto:"ColumnName"`
+	ColumnOrdinal int    `kusto:"ColumnOrdinal"`
+	DataType      string `kusto:"DataType"`
+	ColumnType    string `kusto:"ColumnType"`
+}
+
+func cslSchemaFromKqlSchema(k []*KqlSchema) []string {
+	var s []string
+	for _, col := range k {
+		s = append(s, fmt.Sprintf("%s:%s", col.ColumnName, col.ColumnType))
+	}
+	return s
+}