-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Ingestor support for Kustainer (#448)
Kustainer has several limitations that impact Ingestor's interaction with Kusto. - Kustainer has no auth - Kustainer only supports inline ingestion To support the lack of auth, this change inspects the Kusto cluster URL and if it's an insecure connection, we do not attempt to add auth to the Kusto connection object. To support inline ingestion, we query Kusto's cluster details and if the cluster is marked as "KustoPersonal", we use this as a trigger to switch into an inline ingestion mode.
- Loading branch information
Showing
7 changed files
with
388 additions
and
10 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| package adx | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
|
|
||
| "github.com/Azure/adx-mon/pkg/testutils/kustainer" | ||
| "github.com/Azure/azure-kusto-go/kusto" | ||
| "github.com/stretchr/testify/require" | ||
| "github.com/testcontainers/testcontainers-go" | ||
| ) | ||
|
|
||
| func TestClusterRequiresDirectIngest(t *testing.T) { | ||
| ctx := context.Background() | ||
| k, err := kustainer.Run(ctx, "mcr.microsoft.com/azuredataexplorer/kustainer-linux:latest", kustainer.WithStarted()) | ||
| testcontainers.CleanupContainer(t, k) | ||
| require.NoError(t, err) | ||
|
|
||
| cb := kusto.NewConnectionStringBuilder(k.ConnectionUrl()) | ||
| client, err := kusto.New(cb) | ||
| require.NoError(t, err) | ||
| defer client.Close() | ||
|
|
||
| u := &uploader{ | ||
| KustoCli: client, | ||
| database: "NetDefaultDB", | ||
| } | ||
| requiresDirectIngest, err := u.clusterRequiresDirectIngest(ctx) | ||
| require.NoError(t, err) | ||
| require.True(t, requiresDirectIngest) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,146 @@ | ||
| package testutils | ||
|
|
||
| import ( | ||
| "context" | ||
| "io" | ||
| "testing" | ||
|
|
||
| "github.com/Azure/azure-kusto-go/kusto" | ||
| "github.com/Azure/azure-kusto-go/kusto/kql" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TableExists(ctx context.Context, t *testing.T, database, table, uri string) bool { | ||
| t.Helper() | ||
|
|
||
| cb := kusto.NewConnectionStringBuilder(uri) | ||
| client, err := kusto.New(cb) | ||
| require.NoError(t, err) | ||
| defer client.Close() | ||
|
|
||
| stmt := kql.New(".show tables") | ||
| rows, err := client.Mgmt(ctx, database, stmt) | ||
| require.NoError(t, err) | ||
| defer rows.Stop() | ||
|
|
||
| for { | ||
| row, errInline, errFinal := rows.NextRowOrError() | ||
| if errFinal == io.EOF { | ||
| break | ||
| } | ||
| if errInline != nil { | ||
| t.Logf("Partial failure to retrieve tables: %v", errInline) | ||
| continue | ||
| } | ||
| if errFinal != nil { | ||
| t.Errorf("Failed to retrieve tables: %v", errFinal) | ||
| } | ||
|
|
||
| var tbl Table | ||
| if err := row.ToStruct(&tbl); err != nil { | ||
| t.Errorf("Failed to convert row to struct: %v", err) | ||
| continue | ||
| } | ||
| if tbl.TableName == table { | ||
| return true | ||
| } | ||
| } | ||
|
|
||
| return false | ||
| } | ||
|
|
||
| type Table struct { | ||
| TableName string `kusto:"TableName"` | ||
| DatabaseName string `kusto:"DatabaseName"` | ||
| Folder string `kusto:"Folder"` | ||
| DocString string `kusto:"DocString"` | ||
| } | ||
|
|
||
| func FunctionExists(ctx context.Context, t *testing.T, database, function, uri string) bool { | ||
| t.Helper() | ||
|
|
||
| cb := kusto.NewConnectionStringBuilder(uri) | ||
| client, err := kusto.New(cb) | ||
| require.NoError(t, err) | ||
| defer client.Close() | ||
|
|
||
| stmt := kql.New(".show functions") | ||
| rows, err := client.Mgmt(ctx, database, stmt) | ||
| require.NoError(t, err) | ||
| defer rows.Stop() | ||
|
|
||
| for { | ||
| row, errInline, errFinal := rows.NextRowOrError() | ||
| if errFinal == io.EOF { | ||
| break | ||
| } | ||
| if errInline != nil { | ||
| t.Logf("Partial failure to retrieve functions: %v", errInline) | ||
| continue | ||
| } | ||
| if errFinal != nil { | ||
| t.Errorf("Failed to retrieve functions: %v", errFinal) | ||
| } | ||
|
|
||
| var fn Function | ||
| if err := row.ToStruct(&fn); err != nil { | ||
| t.Errorf("Failed to convert row to struct: %v", err) | ||
| continue | ||
| } | ||
| if fn.Name == function { | ||
| return true | ||
| } | ||
| } | ||
|
|
||
| return false | ||
| } | ||
|
|
||
| type Function struct { | ||
| Name string `kusto:"Name"` | ||
| Parameters string `kusto:"Parameters"` | ||
| Body string `kusto:"Body"` | ||
| Folder string `kusto:"Folder"` | ||
| DocString string `kusto:"DocString"` | ||
| } | ||
|
|
||
| func TableHasRows(ctx context.Context, t *testing.T, database, table, uri string) bool { | ||
| t.Helper() | ||
|
|
||
| cb := kusto.NewConnectionStringBuilder(uri) | ||
| client, err := kusto.New(cb) | ||
| require.NoError(t, err) | ||
| defer client.Close() | ||
|
|
||
| query := kql.New("").AddUnsafe(table).AddLiteral(" | count") | ||
| rows, err := client.Query(ctx, database, query) | ||
| require.NoError(t, err) | ||
| defer rows.Stop() | ||
|
|
||
| for { | ||
| row, errInline, errFinal := rows.NextRowOrError() | ||
| if errFinal == io.EOF { | ||
| break | ||
| } | ||
| if errInline != nil { | ||
| t.Logf("Partial failure to retrieve row count: %v", errInline) | ||
| continue | ||
| } | ||
| if errFinal != nil { | ||
| t.Errorf("Failed to retrieve row count: %v", errFinal) | ||
| } | ||
|
|
||
| var count RowCount | ||
| if err := row.ToStruct(&count); err != nil { | ||
| t.Errorf("Failed to convert row to struct: %v", err) | ||
| continue | ||
| } | ||
| return count.Count > 0 | ||
| } | ||
|
|
||
| t.Logf("Table %s has no rows", table) | ||
| return false | ||
| } | ||
|
|
||
| type RowCount struct { | ||
| Count int64 `kusto:"Count"` | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,28 @@ | ||
| package testutils_test | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
|
|
||
| "github.com/Azure/adx-mon/pkg/testutils" | ||
| "github.com/Azure/adx-mon/pkg/testutils/kustainer" | ||
| "github.com/stretchr/testify/require" | ||
| "github.com/testcontainers/testcontainers-go" | ||
| ) | ||
|
|
||
| func TestTableExists(t *testing.T) { | ||
| k, err := kustainer.Run(context.Background(), "mcr.microsoft.com/azuredataexplorer/kustainer-linux:latest", kustainer.WithStarted()) | ||
| testcontainers.CleanupContainer(t, k) | ||
| require.NoError(t, err) | ||
|
|
||
| require.False(t, testutils.TableExists(context.Background(), t, "NetDefaultDB", "Foo", k.ConnectionUrl())) | ||
| require.True(t, testutils.TableExists(context.Background(), t, "NetDefaultDB", "Table_0", k.ConnectionUrl())) | ||
| } | ||
|
|
||
| func TestTableHasRows(t *testing.T) { | ||
| k, err := kustainer.Run(context.Background(), "mcr.microsoft.com/azuredataexplorer/kustainer-linux:latest", kustainer.WithStarted()) | ||
| testcontainers.CleanupContainer(t, k) | ||
| require.NoError(t, err) | ||
|
|
||
| require.False(t, testutils.TableHasRows(context.Background(), t, "NetDefaultDB", "Table_0", k.ConnectionUrl())) | ||
| } |
Oops, something went wrong.