From 8054a1b550f3e38aaa8208e28c8fad88f6a51d35 Mon Sep 17 00:00:00 2001 From: Ramachandran G Date: Tue, 27 Sep 2022 07:57:39 +0530 Subject: [PATCH 1/5] * Added support for managed ngestion * Address review comments --- plugins/outputs/azure_data_explorer/README.md | 24 +++ .../azure_data_explorer.go | 159 ++++++++++++++++-- .../azure_data_explorer_test.go | 136 +++++++++++++++ .../outputs/azure_data_explorer/sample.conf | 6 + 4 files changed, 311 insertions(+), 14 deletions(-) diff --git a/plugins/outputs/azure_data_explorer/README.md b/plugins/outputs/azure_data_explorer/README.md index 35cc7489ad510..db78a1f0981e1 100644 --- a/plugins/outputs/azure_data_explorer/README.md +++ b/plugins/outputs/azure_data_explorer/README.md @@ -41,8 +41,20 @@ of logs, metrics and time series data. ## Creates tables and relevant mapping if set to true(default). ## Skips table and mapping creation if set to false, this is useful for running Telegraf with the lowest possible permissions i.e. table ingestor role. # create_tables = true + + ## Ingestion method to use. + ## Available options are + ## - managed -- streaming ingestion with fallback to batched ingestion or the "queued" method below + ## - queued -- queue up metrics data and process sequentially + # ingestion_type = "queued" ``` +## Deprecations + +- `localClient` and `localIngestor` has been deprecated since the introduction of [`kusto.Client`](https://github.com/Azure/azure-kusto-go/blob/master/kusto/kusto.go#L111) and [`ingest.Ingestor`](https://github.com/Azure/azure-kusto-go/blob/master/kusto/ingest/ingest.go#L18) respectively in [Auzre Kusto Go SDK](https://github.com/Azure/azure-kusto-go). The change is done to maintain the consistency. +- Because the deprications are internal, changes will not impact public usage, +- The related deprecated code will be removed in the next update. + ## Metrics Grouping Metrics can be grouped in two ways to be sent to Azure Data Explorer. To specify @@ -93,6 +105,18 @@ The corresponding table mapping would be like the following: **Note**: This plugin will automatically create Azure Data Explorer tables and corresponding table mapping as per the above mentioned commands. +## Ingestion type + +**Note**: +[Streaming ingestion](https://aka.ms/AAhlg6s) +has to be enabled on ADX [configure the ADX cluster] +in case of `managed` option. +Refer the query below to check if streaming is enabled + +```kql +.show database policy streamingingestion +``` + ## Authentiation ### Supported Authentication Methods diff --git a/plugins/outputs/azure_data_explorer/azure_data_explorer.go b/plugins/outputs/azure_data_explorer/azure_data_explorer.go index 74b3f69e36f89..aeea08af8c7d7 100644 --- a/plugins/outputs/azure_data_explorer/azure_data_explorer.go +++ b/plugins/outputs/azure_data_explorer/azure_data_explorer.go @@ -12,12 +12,14 @@ import ( "time" "github.com/Azure/azure-kusto-go/kusto" + kustoerrors "github.com/Azure/azure-kusto-go/kusto/data/errors" "github.com/Azure/azure-kusto-go/kusto/ingest" "github.com/Azure/azure-kusto-go/kusto/unsafe" "github.com/Azure/go-autorest/autorest/azure/auth" "github.com/influxdata/telegraf" "github.com/influxdata/telegraf/config" + "github.com/influxdata/telegraf/internal/choice" "github.com/influxdata/telegraf/plugins/outputs" "github.com/influxdata/telegraf/plugins/serializers" "github.com/influxdata/telegraf/plugins/serializers/json" @@ -34,10 +36,17 @@ type AzureDataExplorer struct { MetricsGrouping string `toml:"metrics_grouping_type"` TableName string `toml:"table_name"` CreateTables bool `toml:"create_tables"` - client localClient - ingesters map[string]localIngestor - serializer serializers.Serializer - createIngestor ingestorFactory + //Deprecated: client of type *kusto.Client, ingestors of type ingest.Ingestor introduced + client localClient + ingesters map[string]localIngestor + /***/ + serializer serializers.Serializer + //Deprecated + createIngestor ingestorFactory + /***/ + IngestionType string `toml:"ingestion_type"` + kustoClient *kusto.Client + metricIngestors map[string]ingest.Ingestor } const ( @@ -60,11 +69,14 @@ type ingestorFactory func(localClient, string, string) (localIngestor, error) const createTableCommand = `.create-merge table ['%s'] (['fields']:dynamic, ['name']:string, ['tags']:dynamic, ['timestamp']:datetime);` const createTableMappingCommand = `.create-or-alter table ['%s'] ingestion json mapping '%s_mapping' '[{"column":"fields", "Properties":{"Path":"$[\'fields\']"}},{"column":"name", "Properties":{"Path":"$[\'name\']"}},{"column":"tags", "Properties":{"Path":"$[\'tags\']"}},{"column":"timestamp", "Properties":{"Path":"$[\'timestamp\']"}}]'` +const managedIngestion = "managed" +const queuedIngestion = "queued" func (*AzureDataExplorer) SampleConfig() string { return sampleConfig } +// Initialize the client and the ingestor func (adx *AzureDataExplorer) Connect() error { authorizer, err := auth.NewAuthorizerFromEnvironmentWithResource(adx.Endpoint) if err != nil { @@ -78,18 +90,45 @@ func (adx *AzureDataExplorer) Connect() error { if err != nil { return err } + adx.kustoClient = client + adx.metricIngestors = make(map[string]ingest.Ingestor) + //Depticated adx.client = client adx.ingesters = make(map[string]localIngestor) adx.createIngestor = createRealIngestor + /***/ return nil } +// Clean up and close the ingestor func (adx *AzureDataExplorer) Close() error { + var errs []error + for _, v := range adx.metricIngestors { + if err := v.Close(); err != nil { + // accumulate errors while closing ingestors + errs = append(errs, err) + } + } + if err := adx.kustoClient.Close(); err != nil { + errs = append(errs, err) + } + + adx.kustoClient = nil + adx.metricIngestors = nil + + if len(errs) == 0 { + adx.Log.Info("Closed ingestors and client") + return nil + } + + //Deprecated adx.client = nil adx.ingesters = nil + /***/ - return nil + // Combine errors into a single object and return the combined error + return kustoerrors.GetCombinedError(errs...) } func (adx *AzureDataExplorer) Write(metrics []telegraf.Metric) error { @@ -151,19 +190,58 @@ func (adx *AzureDataExplorer) writeSingleTable(metrics []telegraf.Metric) error } func (adx *AzureDataExplorer) pushMetrics(ctx context.Context, format ingest.FileOption, tableName string, metricsArray []byte) error { - ingestor, err := adx.getIngestor(ctx, tableName) - if err != nil { - return err + var ingestor localIngestor + var metricIngestor ingest.Ingestor + var err error + if adx.client != nil && adx.createIngestor != nil { + ingestor, err = adx.getIngestor(ctx, tableName) + if err != nil { + return err + } + } else { + metricIngestor, err = adx.getMetricIngestor(ctx, tableName) + if err != nil { + return err + } } + length := len(metricsArray) + adx.Log.Debugf("Writing %s metrics to table %q", length, tableName) reader := bytes.NewReader(metricsArray) mapping := ingest.IngestionMappingRef(fmt.Sprintf("%s_mapping", tableName), ingest.JSON) - if _, err := ingestor.FromReader(ctx, reader, format, mapping); err != nil { - adx.Log.Errorf("sending ingestion request to Azure Data Explorer for table %q failed: %v", tableName, err) + if ingestor != nil { + //Deprecated + if _, err := ingestor.FromReader(ctx, reader, format, mapping); err != nil { + adx.Log.Errorf("sending ingestion request to Azure Data Explorer for table %q failed: %v", tableName, err) + } + } else if metricIngestor != nil { + if _, err := metricIngestor.FromReader(ctx, reader, format, mapping); err != nil { + adx.Log.Errorf("sending ingestion request to Azure Data Explorer for table %q failed: %v", tableName, err) + } } return nil } +func (adx *AzureDataExplorer) getMetricIngestor(ctx context.Context, tableName string) (ingest.Ingestor, error) { + ingestor := adx.metricIngestors[tableName] + + if ingestor == nil { + if err := adx.createAzureDataExplorerTable(ctx, tableName); err != nil { + return nil, fmt.Errorf("creating table for %q failed: %v", tableName, err) + } + //create a new ingestor client for the table + tempIngestor, err := createIngestorByTable(adx.kustoClient, adx.Database, tableName, adx.IngestionType) + if err != nil { + return nil, fmt.Errorf("creating ingestor for %q failed: %v", tableName, err) + } + adx.metricIngestors[tableName] = tempIngestor + adx.Log.Debugf("Ingestor for table %s created", tableName) + ingestor = tempIngestor + } + return ingestor, nil +} + +//Deprecated: getMetricIngestor introduced to use inget.Ingestor instead of localIngestor func (adx *AzureDataExplorer) getIngestor(ctx context.Context, tableName string) (localIngestor, error) { ingestor := adx.ingesters[tableName] @@ -182,19 +260,33 @@ func (adx *AzureDataExplorer) getIngestor(ctx context.Context, tableName string) return ingestor, nil } +/***/ + func (adx *AzureDataExplorer) createAzureDataExplorerTable(ctx context.Context, tableName string) error { if !adx.CreateTables { adx.Log.Info("skipped table creation") return nil } createStmt := kusto.NewStmt("", kusto.UnsafeStmt(unsafe.Stmt{Add: true, SuppressWarning: true})).UnsafeAdd(fmt.Sprintf(createTableCommand, tableName)) - if _, err := adx.client.Mgmt(ctx, adx.Database, createStmt); err != nil { - return err + if adx.client != nil { + if _, err := adx.client.Mgmt(ctx, adx.Database, createStmt); err != nil { + return err + } + } else if adx.kustoClient != nil { + if _, err := adx.kustoClient.Mgmt(ctx, adx.Database, createStmt); err != nil { + return err + } } createTableMappingstmt := kusto.NewStmt("", kusto.UnsafeStmt(unsafe.Stmt{Add: true, SuppressWarning: true})).UnsafeAdd(fmt.Sprintf(createTableMappingCommand, tableName, tableName)) - if _, err := adx.client.Mgmt(ctx, adx.Database, createTableMappingstmt); err != nil { - return err + if adx.client != nil { + if _, err := adx.client.Mgmt(ctx, adx.Database, createTableMappingstmt); err != nil { + return err + } + } else if adx.kustoClient != nil { + if _, err := adx.kustoClient.Mgmt(ctx, adx.Database, createTableMappingstmt); err != nil { + return err + } } return nil @@ -219,6 +311,12 @@ func (adx *AzureDataExplorer) Init() error { return errors.New("Metrics grouping type is not valid") } + if adx.IngestionType == "" { + adx.IngestionType = queuedIngestion + } else if !(choice.Contains(adx.IngestionType, []string{managedIngestion, queuedIngestion})) { + return fmt.Errorf("unknown ingestion type %q", adx.IngestionType) + } + serializer, err := json.NewSerializer(time.Nanosecond, time.RFC3339Nano, "") if err != nil { return err @@ -236,6 +334,7 @@ func init() { }) } +//Deprecated: createIngestorByTable should be used with ingestionType and ingest.Ingestor func createRealIngestor(client localClient, database string, tableName string) (localIngestor, error) { ingestor, err := ingest.New(client.(*kusto.Client), database, tableName, ingest.WithStaticBuffer(bufferSize, maxBuffers)) if ingestor != nil { @@ -243,3 +342,35 @@ func createRealIngestor(client localClient, database string, tableName string) ( } return nil, err } + +// For each table create the ingestor +func createIngestorByTable(client *kusto.Client, database string, tableName string, ingestionType string) (ingest.Ingestor, error) { + var ingestor ingest.Ingestor + var err error + switch strings.ToLower(ingestionType) { + case managedIngestion: + { + mi, err := ingest.NewManaged(client, database, tableName) + if err != nil { + return nil, err + } + ingestor = mi + } + case queuedIngestion: + { + qi, err := ingest.New(client, database, tableName, ingest.WithStaticBuffer(bufferSize, maxBuffers)) + if err != nil { + return nil, err + } + ingestor = qi + } + default: + { + err = fmt.Errorf(`ingestion_type has to be one of %q or %q`, managedIngestion, queuedIngestion) + } + } + if ingestor != nil { + return ingestor, nil + } + return nil, err +} diff --git a/plugins/outputs/azure_data_explorer/azure_data_explorer_test.go b/plugins/outputs/azure_data_explorer/azure_data_explorer_test.go index 83a09fa4f43f3..b4fe34f30ebf9 100644 --- a/plugins/outputs/azure_data_explorer/azure_data_explorer_test.go +++ b/plugins/outputs/azure_data_explorer/azure_data_explorer_test.go @@ -7,6 +7,8 @@ import ( "errors" "fmt" "io" + "io/ioutil" + "strings" "testing" "time" @@ -21,6 +23,7 @@ import ( const createTableCommandExpected = `.create-merge table ['%s'] (['fields']:dynamic, ['name']:string, ['tags']:dynamic, ['timestamp']:datetime);` const createTableMappingCommandExpected = `.create-or-alter table ['%s'] ingestion json mapping '%s_mapping' '[{"column":"fields", "Properties":{"Path":"$[\'fields\']"}},{"column":"name", "Properties":{"Path":"$[\'name\']"}},{"column":"tags", "Properties":{"Path":"$[\'tags\']"}},{"column":"timestamp", "Properties":{"Path":"$[\'timestamp\']"}}]'` +//Deprecated func TestWrite(t *testing.T) { testCases := []struct { name string @@ -193,6 +196,111 @@ func TestWrite(t *testing.T) { } } +/***/ + +func TestWriteWithType(t *testing.T) { + metricName := "test1" + fakeClient := kusto.NewMockClient() + expectedResultMap := map[string]string{metricName: `{"fields":{"value":1},"name":"test1","tags":{"tag1":"value1"},"timestamp":1257894000}`} + mockMetrics := testutil.MockMetrics() + // Multi tables + mockMetrics2 := testutil.TestMetric(1.0, "test2") + mockMetrics3 := testutil.TestMetric(2.0, "test3") + mockMetricsMulti := make([]telegraf.Metric, 2) + mockMetricsMulti[0] = mockMetrics2 + mockMetricsMulti[1] = mockMetrics3 + expectedResultMap2 := map[string]string{"test2": `{"fields":{"value":1.0},"name":"test2","tags":{"tag1":"value1"},"timestamp":1257894000}`, "test3": `{"fields":{"value":2.0},"name":"test3","tags":{"tag1":"value1"},"timestamp":1257894000}`} + // List of tests + testCases := []struct { + name string + inputMetric []telegraf.Metric + metricsGrouping string + tableNameToExpectedResult map[string]string + expectedWriteError string + createTables bool + ingestionType string + }{ + { + name: "Valid metric", + inputMetric: mockMetrics, + createTables: true, + metricsGrouping: tablePerMetric, + tableNameToExpectedResult: expectedResultMap, + }, + { + name: "Don't create tables'", + inputMetric: mockMetrics, + createTables: false, + metricsGrouping: tablePerMetric, + tableNameToExpectedResult: expectedResultMap, + }, + { + name: "SingleTable metric grouping type", + inputMetric: mockMetrics, + createTables: true, + metricsGrouping: singleTable, + tableNameToExpectedResult: expectedResultMap, + }, + { + name: "Valid metric managed ingestion", + inputMetric: mockMetrics, + createTables: true, + metricsGrouping: tablePerMetric, + tableNameToExpectedResult: expectedResultMap, + ingestionType: managedIngestion, + }, + { + name: "Table per metric type", + inputMetric: mockMetricsMulti, + createTables: true, + metricsGrouping: tablePerMetric, + tableNameToExpectedResult: expectedResultMap2, + }, + } + for _, testCase := range testCases { + testCase := testCase + t.Run(testCase.name, func(t *testing.T) { + //t.Parallel() + serializer, err := telegrafJson.NewSerializer(time.Second, "", "") + require.NoError(t, err) + for tableName, jsonValue := range testCase.tableNameToExpectedResult { + ingestionType := "queued" + if testCase.ingestionType != "" { + ingestionType = testCase.ingestionType + } + mockIngestor := &mockIngestor{} + plugin := AzureDataExplorer{ + Endpoint: "someendpoint", + Database: "databasename", + Log: testutil.Logger{}, + IngestionType: ingestionType, + MetricsGrouping: testCase.metricsGrouping, + TableName: tableName, + CreateTables: testCase.createTables, + kustoClient: fakeClient, + metricIngestors: map[string]ingest.Ingestor{ + tableName: mockIngestor, + }, + serializer: serializer, + } + errorInWrite := plugin.Write(testCase.inputMetric) + if testCase.expectedWriteError != "" { + require.EqualError(t, errorInWrite, testCase.expectedWriteError) + } else { + require.NoError(t, errorInWrite) + createdIngestor := plugin.metricIngestors[tableName] + if testCase.metricsGrouping == singleTable { + createdIngestor = plugin.metricIngestors[tableName] + } + records := mockIngestor.records[0] // the first element + require.NotNil(t, createdIngestor) + require.JSONEq(t, jsonValue, records) + } + } + }) + } +} + func TestInitBlankEndpoint(t *testing.T) { plugin := AzureDataExplorer{ Log: testutil.Logger{}, @@ -232,3 +340,31 @@ func (f *fakeIngestor) FromReader(_ context.Context, reader io.Reader, _ ...inge } return &ingest.Result{}, nil } + +type mockIngestor struct { + records []string +} + +func (m *mockIngestor) FromReader(ctx context.Context, reader io.Reader, options ...ingest.FileOption) (*ingest.Result, error) { + bufbytes, _ := ioutil.ReadAll(reader) + metricjson := string(bufbytes) + m.SetRecords(strings.Split(metricjson, "\n")) + return &ingest.Result{}, nil +} + +func (m *mockIngestor) FromFile(ctx context.Context, fPath string, options ...ingest.FileOption) (*ingest.Result, error) { + return &ingest.Result{}, nil +} + +func (m *mockIngestor) SetRecords(records []string) { + m.records = records +} + +// Name receives a copy of Foo since it doesn't need to modify it. +func (m *mockIngestor) Records() []string { + return m.records +} + +func (m *mockIngestor) Close() error { + return nil +} diff --git a/plugins/outputs/azure_data_explorer/sample.conf b/plugins/outputs/azure_data_explorer/sample.conf index 5f8965b30fe59..f708c7808e3ee 100644 --- a/plugins/outputs/azure_data_explorer/sample.conf +++ b/plugins/outputs/azure_data_explorer/sample.conf @@ -23,3 +23,9 @@ ## Creates tables and relevant mapping if set to true(default). ## Skips table and mapping creation if set to false, this is useful for running Telegraf with the lowest possible permissions i.e. table ingestor role. # create_tables = true + + ## Ingestion method to use. + ## Available options are + ## - managed -- streaming ingestion with fallback to batched ingestion or the "queued" method below + ## - queued -- queue up metrics data and process sequentially + # ingestion_type = "queued" From 082b4740442eaa646570b78433bdeee3a8a849a6 Mon Sep 17 00:00:00 2001 From: Ramachandran G Date: Tue, 27 Sep 2022 07:57:39 +0530 Subject: [PATCH 2/5] feat: Add support for managed streaming ingestion --- plugins/outputs/azure_data_explorer/README.md | 24 +++ .../azure_data_explorer.go | 159 ++++++++++++++++-- .../azure_data_explorer_test.go | 136 +++++++++++++++ .../outputs/azure_data_explorer/sample.conf | 6 + 4 files changed, 311 insertions(+), 14 deletions(-) diff --git a/plugins/outputs/azure_data_explorer/README.md b/plugins/outputs/azure_data_explorer/README.md index 35cc7489ad510..db78a1f0981e1 100644 --- a/plugins/outputs/azure_data_explorer/README.md +++ b/plugins/outputs/azure_data_explorer/README.md @@ -41,8 +41,20 @@ of logs, metrics and time series data. ## Creates tables and relevant mapping if set to true(default). ## Skips table and mapping creation if set to false, this is useful for running Telegraf with the lowest possible permissions i.e. table ingestor role. # create_tables = true + + ## Ingestion method to use. + ## Available options are + ## - managed -- streaming ingestion with fallback to batched ingestion or the "queued" method below + ## - queued -- queue up metrics data and process sequentially + # ingestion_type = "queued" ``` +## Deprecations + +- `localClient` and `localIngestor` has been deprecated since the introduction of [`kusto.Client`](https://github.com/Azure/azure-kusto-go/blob/master/kusto/kusto.go#L111) and [`ingest.Ingestor`](https://github.com/Azure/azure-kusto-go/blob/master/kusto/ingest/ingest.go#L18) respectively in [Auzre Kusto Go SDK](https://github.com/Azure/azure-kusto-go). The change is done to maintain the consistency. +- Because the deprications are internal, changes will not impact public usage, +- The related deprecated code will be removed in the next update. + ## Metrics Grouping Metrics can be grouped in two ways to be sent to Azure Data Explorer. To specify @@ -93,6 +105,18 @@ The corresponding table mapping would be like the following: **Note**: This plugin will automatically create Azure Data Explorer tables and corresponding table mapping as per the above mentioned commands. +## Ingestion type + +**Note**: +[Streaming ingestion](https://aka.ms/AAhlg6s) +has to be enabled on ADX [configure the ADX cluster] +in case of `managed` option. +Refer the query below to check if streaming is enabled + +```kql +.show database policy streamingingestion +``` + ## Authentiation ### Supported Authentication Methods diff --git a/plugins/outputs/azure_data_explorer/azure_data_explorer.go b/plugins/outputs/azure_data_explorer/azure_data_explorer.go index 74b3f69e36f89..aeea08af8c7d7 100644 --- a/plugins/outputs/azure_data_explorer/azure_data_explorer.go +++ b/plugins/outputs/azure_data_explorer/azure_data_explorer.go @@ -12,12 +12,14 @@ import ( "time" "github.com/Azure/azure-kusto-go/kusto" + kustoerrors "github.com/Azure/azure-kusto-go/kusto/data/errors" "github.com/Azure/azure-kusto-go/kusto/ingest" "github.com/Azure/azure-kusto-go/kusto/unsafe" "github.com/Azure/go-autorest/autorest/azure/auth" "github.com/influxdata/telegraf" "github.com/influxdata/telegraf/config" + "github.com/influxdata/telegraf/internal/choice" "github.com/influxdata/telegraf/plugins/outputs" "github.com/influxdata/telegraf/plugins/serializers" "github.com/influxdata/telegraf/plugins/serializers/json" @@ -34,10 +36,17 @@ type AzureDataExplorer struct { MetricsGrouping string `toml:"metrics_grouping_type"` TableName string `toml:"table_name"` CreateTables bool `toml:"create_tables"` - client localClient - ingesters map[string]localIngestor - serializer serializers.Serializer - createIngestor ingestorFactory + //Deprecated: client of type *kusto.Client, ingestors of type ingest.Ingestor introduced + client localClient + ingesters map[string]localIngestor + /***/ + serializer serializers.Serializer + //Deprecated + createIngestor ingestorFactory + /***/ + IngestionType string `toml:"ingestion_type"` + kustoClient *kusto.Client + metricIngestors map[string]ingest.Ingestor } const ( @@ -60,11 +69,14 @@ type ingestorFactory func(localClient, string, string) (localIngestor, error) const createTableCommand = `.create-merge table ['%s'] (['fields']:dynamic, ['name']:string, ['tags']:dynamic, ['timestamp']:datetime);` const createTableMappingCommand = `.create-or-alter table ['%s'] ingestion json mapping '%s_mapping' '[{"column":"fields", "Properties":{"Path":"$[\'fields\']"}},{"column":"name", "Properties":{"Path":"$[\'name\']"}},{"column":"tags", "Properties":{"Path":"$[\'tags\']"}},{"column":"timestamp", "Properties":{"Path":"$[\'timestamp\']"}}]'` +const managedIngestion = "managed" +const queuedIngestion = "queued" func (*AzureDataExplorer) SampleConfig() string { return sampleConfig } +// Initialize the client and the ingestor func (adx *AzureDataExplorer) Connect() error { authorizer, err := auth.NewAuthorizerFromEnvironmentWithResource(adx.Endpoint) if err != nil { @@ -78,18 +90,45 @@ func (adx *AzureDataExplorer) Connect() error { if err != nil { return err } + adx.kustoClient = client + adx.metricIngestors = make(map[string]ingest.Ingestor) + //Depticated adx.client = client adx.ingesters = make(map[string]localIngestor) adx.createIngestor = createRealIngestor + /***/ return nil } +// Clean up and close the ingestor func (adx *AzureDataExplorer) Close() error { + var errs []error + for _, v := range adx.metricIngestors { + if err := v.Close(); err != nil { + // accumulate errors while closing ingestors + errs = append(errs, err) + } + } + if err := adx.kustoClient.Close(); err != nil { + errs = append(errs, err) + } + + adx.kustoClient = nil + adx.metricIngestors = nil + + if len(errs) == 0 { + adx.Log.Info("Closed ingestors and client") + return nil + } + + //Deprecated adx.client = nil adx.ingesters = nil + /***/ - return nil + // Combine errors into a single object and return the combined error + return kustoerrors.GetCombinedError(errs...) } func (adx *AzureDataExplorer) Write(metrics []telegraf.Metric) error { @@ -151,19 +190,58 @@ func (adx *AzureDataExplorer) writeSingleTable(metrics []telegraf.Metric) error } func (adx *AzureDataExplorer) pushMetrics(ctx context.Context, format ingest.FileOption, tableName string, metricsArray []byte) error { - ingestor, err := adx.getIngestor(ctx, tableName) - if err != nil { - return err + var ingestor localIngestor + var metricIngestor ingest.Ingestor + var err error + if adx.client != nil && adx.createIngestor != nil { + ingestor, err = adx.getIngestor(ctx, tableName) + if err != nil { + return err + } + } else { + metricIngestor, err = adx.getMetricIngestor(ctx, tableName) + if err != nil { + return err + } } + length := len(metricsArray) + adx.Log.Debugf("Writing %s metrics to table %q", length, tableName) reader := bytes.NewReader(metricsArray) mapping := ingest.IngestionMappingRef(fmt.Sprintf("%s_mapping", tableName), ingest.JSON) - if _, err := ingestor.FromReader(ctx, reader, format, mapping); err != nil { - adx.Log.Errorf("sending ingestion request to Azure Data Explorer for table %q failed: %v", tableName, err) + if ingestor != nil { + //Deprecated + if _, err := ingestor.FromReader(ctx, reader, format, mapping); err != nil { + adx.Log.Errorf("sending ingestion request to Azure Data Explorer for table %q failed: %v", tableName, err) + } + } else if metricIngestor != nil { + if _, err := metricIngestor.FromReader(ctx, reader, format, mapping); err != nil { + adx.Log.Errorf("sending ingestion request to Azure Data Explorer for table %q failed: %v", tableName, err) + } } return nil } +func (adx *AzureDataExplorer) getMetricIngestor(ctx context.Context, tableName string) (ingest.Ingestor, error) { + ingestor := adx.metricIngestors[tableName] + + if ingestor == nil { + if err := adx.createAzureDataExplorerTable(ctx, tableName); err != nil { + return nil, fmt.Errorf("creating table for %q failed: %v", tableName, err) + } + //create a new ingestor client for the table + tempIngestor, err := createIngestorByTable(adx.kustoClient, adx.Database, tableName, adx.IngestionType) + if err != nil { + return nil, fmt.Errorf("creating ingestor for %q failed: %v", tableName, err) + } + adx.metricIngestors[tableName] = tempIngestor + adx.Log.Debugf("Ingestor for table %s created", tableName) + ingestor = tempIngestor + } + return ingestor, nil +} + +//Deprecated: getMetricIngestor introduced to use inget.Ingestor instead of localIngestor func (adx *AzureDataExplorer) getIngestor(ctx context.Context, tableName string) (localIngestor, error) { ingestor := adx.ingesters[tableName] @@ -182,19 +260,33 @@ func (adx *AzureDataExplorer) getIngestor(ctx context.Context, tableName string) return ingestor, nil } +/***/ + func (adx *AzureDataExplorer) createAzureDataExplorerTable(ctx context.Context, tableName string) error { if !adx.CreateTables { adx.Log.Info("skipped table creation") return nil } createStmt := kusto.NewStmt("", kusto.UnsafeStmt(unsafe.Stmt{Add: true, SuppressWarning: true})).UnsafeAdd(fmt.Sprintf(createTableCommand, tableName)) - if _, err := adx.client.Mgmt(ctx, adx.Database, createStmt); err != nil { - return err + if adx.client != nil { + if _, err := adx.client.Mgmt(ctx, adx.Database, createStmt); err != nil { + return err + } + } else if adx.kustoClient != nil { + if _, err := adx.kustoClient.Mgmt(ctx, adx.Database, createStmt); err != nil { + return err + } } createTableMappingstmt := kusto.NewStmt("", kusto.UnsafeStmt(unsafe.Stmt{Add: true, SuppressWarning: true})).UnsafeAdd(fmt.Sprintf(createTableMappingCommand, tableName, tableName)) - if _, err := adx.client.Mgmt(ctx, adx.Database, createTableMappingstmt); err != nil { - return err + if adx.client != nil { + if _, err := adx.client.Mgmt(ctx, adx.Database, createTableMappingstmt); err != nil { + return err + } + } else if adx.kustoClient != nil { + if _, err := adx.kustoClient.Mgmt(ctx, adx.Database, createTableMappingstmt); err != nil { + return err + } } return nil @@ -219,6 +311,12 @@ func (adx *AzureDataExplorer) Init() error { return errors.New("Metrics grouping type is not valid") } + if adx.IngestionType == "" { + adx.IngestionType = queuedIngestion + } else if !(choice.Contains(adx.IngestionType, []string{managedIngestion, queuedIngestion})) { + return fmt.Errorf("unknown ingestion type %q", adx.IngestionType) + } + serializer, err := json.NewSerializer(time.Nanosecond, time.RFC3339Nano, "") if err != nil { return err @@ -236,6 +334,7 @@ func init() { }) } +//Deprecated: createIngestorByTable should be used with ingestionType and ingest.Ingestor func createRealIngestor(client localClient, database string, tableName string) (localIngestor, error) { ingestor, err := ingest.New(client.(*kusto.Client), database, tableName, ingest.WithStaticBuffer(bufferSize, maxBuffers)) if ingestor != nil { @@ -243,3 +342,35 @@ func createRealIngestor(client localClient, database string, tableName string) ( } return nil, err } + +// For each table create the ingestor +func createIngestorByTable(client *kusto.Client, database string, tableName string, ingestionType string) (ingest.Ingestor, error) { + var ingestor ingest.Ingestor + var err error + switch strings.ToLower(ingestionType) { + case managedIngestion: + { + mi, err := ingest.NewManaged(client, database, tableName) + if err != nil { + return nil, err + } + ingestor = mi + } + case queuedIngestion: + { + qi, err := ingest.New(client, database, tableName, ingest.WithStaticBuffer(bufferSize, maxBuffers)) + if err != nil { + return nil, err + } + ingestor = qi + } + default: + { + err = fmt.Errorf(`ingestion_type has to be one of %q or %q`, managedIngestion, queuedIngestion) + } + } + if ingestor != nil { + return ingestor, nil + } + return nil, err +} diff --git a/plugins/outputs/azure_data_explorer/azure_data_explorer_test.go b/plugins/outputs/azure_data_explorer/azure_data_explorer_test.go index 83a09fa4f43f3..b4fe34f30ebf9 100644 --- a/plugins/outputs/azure_data_explorer/azure_data_explorer_test.go +++ b/plugins/outputs/azure_data_explorer/azure_data_explorer_test.go @@ -7,6 +7,8 @@ import ( "errors" "fmt" "io" + "io/ioutil" + "strings" "testing" "time" @@ -21,6 +23,7 @@ import ( const createTableCommandExpected = `.create-merge table ['%s'] (['fields']:dynamic, ['name']:string, ['tags']:dynamic, ['timestamp']:datetime);` const createTableMappingCommandExpected = `.create-or-alter table ['%s'] ingestion json mapping '%s_mapping' '[{"column":"fields", "Properties":{"Path":"$[\'fields\']"}},{"column":"name", "Properties":{"Path":"$[\'name\']"}},{"column":"tags", "Properties":{"Path":"$[\'tags\']"}},{"column":"timestamp", "Properties":{"Path":"$[\'timestamp\']"}}]'` +//Deprecated func TestWrite(t *testing.T) { testCases := []struct { name string @@ -193,6 +196,111 @@ func TestWrite(t *testing.T) { } } +/***/ + +func TestWriteWithType(t *testing.T) { + metricName := "test1" + fakeClient := kusto.NewMockClient() + expectedResultMap := map[string]string{metricName: `{"fields":{"value":1},"name":"test1","tags":{"tag1":"value1"},"timestamp":1257894000}`} + mockMetrics := testutil.MockMetrics() + // Multi tables + mockMetrics2 := testutil.TestMetric(1.0, "test2") + mockMetrics3 := testutil.TestMetric(2.0, "test3") + mockMetricsMulti := make([]telegraf.Metric, 2) + mockMetricsMulti[0] = mockMetrics2 + mockMetricsMulti[1] = mockMetrics3 + expectedResultMap2 := map[string]string{"test2": `{"fields":{"value":1.0},"name":"test2","tags":{"tag1":"value1"},"timestamp":1257894000}`, "test3": `{"fields":{"value":2.0},"name":"test3","tags":{"tag1":"value1"},"timestamp":1257894000}`} + // List of tests + testCases := []struct { + name string + inputMetric []telegraf.Metric + metricsGrouping string + tableNameToExpectedResult map[string]string + expectedWriteError string + createTables bool + ingestionType string + }{ + { + name: "Valid metric", + inputMetric: mockMetrics, + createTables: true, + metricsGrouping: tablePerMetric, + tableNameToExpectedResult: expectedResultMap, + }, + { + name: "Don't create tables'", + inputMetric: mockMetrics, + createTables: false, + metricsGrouping: tablePerMetric, + tableNameToExpectedResult: expectedResultMap, + }, + { + name: "SingleTable metric grouping type", + inputMetric: mockMetrics, + createTables: true, + metricsGrouping: singleTable, + tableNameToExpectedResult: expectedResultMap, + }, + { + name: "Valid metric managed ingestion", + inputMetric: mockMetrics, + createTables: true, + metricsGrouping: tablePerMetric, + tableNameToExpectedResult: expectedResultMap, + ingestionType: managedIngestion, + }, + { + name: "Table per metric type", + inputMetric: mockMetricsMulti, + createTables: true, + metricsGrouping: tablePerMetric, + tableNameToExpectedResult: expectedResultMap2, + }, + } + for _, testCase := range testCases { + testCase := testCase + t.Run(testCase.name, func(t *testing.T) { + //t.Parallel() + serializer, err := telegrafJson.NewSerializer(time.Second, "", "") + require.NoError(t, err) + for tableName, jsonValue := range testCase.tableNameToExpectedResult { + ingestionType := "queued" + if testCase.ingestionType != "" { + ingestionType = testCase.ingestionType + } + mockIngestor := &mockIngestor{} + plugin := AzureDataExplorer{ + Endpoint: "someendpoint", + Database: "databasename", + Log: testutil.Logger{}, + IngestionType: ingestionType, + MetricsGrouping: testCase.metricsGrouping, + TableName: tableName, + CreateTables: testCase.createTables, + kustoClient: fakeClient, + metricIngestors: map[string]ingest.Ingestor{ + tableName: mockIngestor, + }, + serializer: serializer, + } + errorInWrite := plugin.Write(testCase.inputMetric) + if testCase.expectedWriteError != "" { + require.EqualError(t, errorInWrite, testCase.expectedWriteError) + } else { + require.NoError(t, errorInWrite) + createdIngestor := plugin.metricIngestors[tableName] + if testCase.metricsGrouping == singleTable { + createdIngestor = plugin.metricIngestors[tableName] + } + records := mockIngestor.records[0] // the first element + require.NotNil(t, createdIngestor) + require.JSONEq(t, jsonValue, records) + } + } + }) + } +} + func TestInitBlankEndpoint(t *testing.T) { plugin := AzureDataExplorer{ Log: testutil.Logger{}, @@ -232,3 +340,31 @@ func (f *fakeIngestor) FromReader(_ context.Context, reader io.Reader, _ ...inge } return &ingest.Result{}, nil } + +type mockIngestor struct { + records []string +} + +func (m *mockIngestor) FromReader(ctx context.Context, reader io.Reader, options ...ingest.FileOption) (*ingest.Result, error) { + bufbytes, _ := ioutil.ReadAll(reader) + metricjson := string(bufbytes) + m.SetRecords(strings.Split(metricjson, "\n")) + return &ingest.Result{}, nil +} + +func (m *mockIngestor) FromFile(ctx context.Context, fPath string, options ...ingest.FileOption) (*ingest.Result, error) { + return &ingest.Result{}, nil +} + +func (m *mockIngestor) SetRecords(records []string) { + m.records = records +} + +// Name receives a copy of Foo since it doesn't need to modify it. +func (m *mockIngestor) Records() []string { + return m.records +} + +func (m *mockIngestor) Close() error { + return nil +} diff --git a/plugins/outputs/azure_data_explorer/sample.conf b/plugins/outputs/azure_data_explorer/sample.conf index 5f8965b30fe59..f708c7808e3ee 100644 --- a/plugins/outputs/azure_data_explorer/sample.conf +++ b/plugins/outputs/azure_data_explorer/sample.conf @@ -23,3 +23,9 @@ ## Creates tables and relevant mapping if set to true(default). ## Skips table and mapping creation if set to false, this is useful for running Telegraf with the lowest possible permissions i.e. table ingestor role. # create_tables = true + + ## Ingestion method to use. + ## Available options are + ## - managed -- streaming ingestion with fallback to batched ingestion or the "queued" method below + ## - queued -- queue up metrics data and process sequentially + # ingestion_type = "queued" From b13bc01e79d2c579c67abfacb564b9eb28dc0a60 Mon Sep 17 00:00:00 2001 From: Abhishek Saharn Date: Tue, 27 Sep 2022 11:34:31 +0530 Subject: [PATCH 3/5] *Resolved comments --- .../azure_data_explorer.go | 31 +++------------- .../azure_data_explorer_test.go | 37 +++++++++---------- 2 files changed, 23 insertions(+), 45 deletions(-) diff --git a/plugins/outputs/azure_data_explorer/azure_data_explorer.go b/plugins/outputs/azure_data_explorer/azure_data_explorer.go index aeea08af8c7d7..40adf7eace796 100644 --- a/plugins/outputs/azure_data_explorer/azure_data_explorer.go +++ b/plugins/outputs/azure_data_explorer/azure_data_explorer.go @@ -36,6 +36,7 @@ type AzureDataExplorer struct { MetricsGrouping string `toml:"metrics_grouping_type"` TableName string `toml:"table_name"` CreateTables bool `toml:"create_tables"` + IngestionType string `toml:"ingestion_type"` //Deprecated: client of type *kusto.Client, ingestors of type ingest.Ingestor introduced client localClient ingesters map[string]localIngestor @@ -44,7 +45,6 @@ type AzureDataExplorer struct { //Deprecated createIngestor ingestorFactory /***/ - IngestionType string `toml:"ingestion_type"` kustoClient *kusto.Client metricIngestors map[string]ingest.Ingestor } @@ -345,32 +345,13 @@ func createRealIngestor(client localClient, database string, tableName string) ( // For each table create the ingestor func createIngestorByTable(client *kusto.Client, database string, tableName string, ingestionType string) (ingest.Ingestor, error) { - var ingestor ingest.Ingestor - var err error switch strings.ToLower(ingestionType) { case managedIngestion: - { - mi, err := ingest.NewManaged(client, database, tableName) - if err != nil { - return nil, err - } - ingestor = mi - } + mi, err := ingest.NewManaged(client, database, tableName) + return mi, err case queuedIngestion: - { - qi, err := ingest.New(client, database, tableName, ingest.WithStaticBuffer(bufferSize, maxBuffers)) - if err != nil { - return nil, err - } - ingestor = qi - } - default: - { - err = fmt.Errorf(`ingestion_type has to be one of %q or %q`, managedIngestion, queuedIngestion) - } + qi, err := ingest.New(client, database, tableName, ingest.WithStaticBuffer(bufferSize, maxBuffers)) + return qi, err } - if ingestor != nil { - return ingestor, nil - } - return nil, err + return nil, fmt.Errorf(`ingestion_type has to be one of %q or %q`, managedIngestion, queuedIngestion) } diff --git a/plugins/outputs/azure_data_explorer/azure_data_explorer_test.go b/plugins/outputs/azure_data_explorer/azure_data_explorer_test.go index b4fe34f30ebf9..7bb2b5c160add 100644 --- a/plugins/outputs/azure_data_explorer/azure_data_explorer_test.go +++ b/plugins/outputs/azure_data_explorer/azure_data_explorer_test.go @@ -7,7 +7,6 @@ import ( "errors" "fmt" "io" - "io/ioutil" "strings" "testing" "time" @@ -23,7 +22,6 @@ import ( const createTableCommandExpected = `.create-merge table ['%s'] (['fields']:dynamic, ['name']:string, ['tags']:dynamic, ['timestamp']:datetime);` const createTableMappingCommandExpected = `.create-or-alter table ['%s'] ingestion json mapping '%s_mapping' '[{"column":"fields", "Properties":{"Path":"$[\'fields\']"}},{"column":"name", "Properties":{"Path":"$[\'name\']"}},{"column":"tags", "Properties":{"Path":"$[\'tags\']"}},{"column":"timestamp", "Properties":{"Path":"$[\'timestamp\']"}}]'` -//Deprecated func TestWrite(t *testing.T) { testCases := []struct { name string @@ -204,11 +202,10 @@ func TestWriteWithType(t *testing.T) { expectedResultMap := map[string]string{metricName: `{"fields":{"value":1},"name":"test1","tags":{"tag1":"value1"},"timestamp":1257894000}`} mockMetrics := testutil.MockMetrics() // Multi tables - mockMetrics2 := testutil.TestMetric(1.0, "test2") - mockMetrics3 := testutil.TestMetric(2.0, "test3") - mockMetricsMulti := make([]telegraf.Metric, 2) - mockMetricsMulti[0] = mockMetrics2 - mockMetricsMulti[1] = mockMetrics3 + mockMetricsMulti := []telegraf.Metric{ + testutil.TestMetric(1.0, "test2"), + testutil.TestMetric(2.0, "test3"), + } expectedResultMap2 := map[string]string{"test2": `{"fields":{"value":1.0},"name":"test2","tags":{"tag1":"value1"},"timestamp":1257894000}`, "test3": `{"fields":{"value":2.0},"name":"test3","tags":{"tag1":"value1"},"timestamp":1257894000}`} // List of tests testCases := []struct { @@ -260,7 +257,7 @@ func TestWriteWithType(t *testing.T) { for _, testCase := range testCases { testCase := testCase t.Run(testCase.name, func(t *testing.T) { - //t.Parallel() + t.Parallel() serializer, err := telegrafJson.NewSerializer(time.Second, "", "") require.NoError(t, err) for tableName, jsonValue := range testCase.tableNameToExpectedResult { @@ -283,19 +280,19 @@ func TestWriteWithType(t *testing.T) { }, serializer: serializer, } - errorInWrite := plugin.Write(testCase.inputMetric) + err := plugin.Write(testCase.inputMetric) if testCase.expectedWriteError != "" { - require.EqualError(t, errorInWrite, testCase.expectedWriteError) - } else { - require.NoError(t, errorInWrite) - createdIngestor := plugin.metricIngestors[tableName] - if testCase.metricsGrouping == singleTable { - createdIngestor = plugin.metricIngestors[tableName] - } - records := mockIngestor.records[0] // the first element - require.NotNil(t, createdIngestor) - require.JSONEq(t, jsonValue, records) + require.EqualError(t, err, testCase.expectedWriteError) + continue } + require.NoError(t, err) + createdIngestor := plugin.metricIngestors[tableName] + if testCase.metricsGrouping == singleTable { + createdIngestor = plugin.metricIngestors[tableName] + } + records := mockIngestor.records[0] // the first element + require.NotNil(t, createdIngestor) + require.JSONEq(t, jsonValue, records) } }) } @@ -346,7 +343,7 @@ type mockIngestor struct { } func (m *mockIngestor) FromReader(ctx context.Context, reader io.Reader, options ...ingest.FileOption) (*ingest.Result, error) { - bufbytes, _ := ioutil.ReadAll(reader) + bufbytes, _ := io.ReadAll(reader) metricjson := string(bufbytes) m.SetRecords(strings.Split(metricjson, "\n")) return &ingest.Result{}, nil From a78d3ed4db7be3c4ba6cf8f18f877f7030977f00 Mon Sep 17 00:00:00 2001 From: Abhishek Saharn Date: Tue, 27 Sep 2022 12:05:25 +0530 Subject: [PATCH 4/5] Removed deprecations section from readme --- plugins/outputs/azure_data_explorer/README.md | 6 ------ 1 file changed, 6 deletions(-) diff --git a/plugins/outputs/azure_data_explorer/README.md b/plugins/outputs/azure_data_explorer/README.md index db78a1f0981e1..b292619351200 100644 --- a/plugins/outputs/azure_data_explorer/README.md +++ b/plugins/outputs/azure_data_explorer/README.md @@ -49,12 +49,6 @@ of logs, metrics and time series data. # ingestion_type = "queued" ``` -## Deprecations - -- `localClient` and `localIngestor` has been deprecated since the introduction of [`kusto.Client`](https://github.com/Azure/azure-kusto-go/blob/master/kusto/kusto.go#L111) and [`ingest.Ingestor`](https://github.com/Azure/azure-kusto-go/blob/master/kusto/ingest/ingest.go#L18) respectively in [Auzre Kusto Go SDK](https://github.com/Azure/azure-kusto-go). The change is done to maintain the consistency. -- Because the deprications are internal, changes will not impact public usage, -- The related deprecated code will be removed in the next update. - ## Metrics Grouping Metrics can be grouped in two ways to be sent to Azure Data Explorer. To specify From 01ad765747e8d64de0ccf864b6af0450a0a0b00b Mon Sep 17 00:00:00 2001 From: Abhishek Saharn Date: Tue, 27 Sep 2022 17:24:57 +0530 Subject: [PATCH 5/5] *Removed test paralleli execution *Lint fix --- plugins/outputs/azure_data_explorer/azure_data_explorer.go | 4 ++-- .../outputs/azure_data_explorer/azure_data_explorer_test.go | 2 -- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/plugins/outputs/azure_data_explorer/azure_data_explorer.go b/plugins/outputs/azure_data_explorer/azure_data_explorer.go index 40adf7eace796..12500709ebc82 100644 --- a/plugins/outputs/azure_data_explorer/azure_data_explorer.go +++ b/plugins/outputs/azure_data_explorer/azure_data_explorer.go @@ -241,7 +241,7 @@ func (adx *AzureDataExplorer) getMetricIngestor(ctx context.Context, tableName s return ingestor, nil } -//Deprecated: getMetricIngestor introduced to use inget.Ingestor instead of localIngestor +// Deprecated: getMetricIngestor introduced to use inget.Ingestor instead of localIngestor func (adx *AzureDataExplorer) getIngestor(ctx context.Context, tableName string) (localIngestor, error) { ingestor := adx.ingesters[tableName] @@ -334,7 +334,7 @@ func init() { }) } -//Deprecated: createIngestorByTable should be used with ingestionType and ingest.Ingestor +// Deprecated: createIngestorByTable should be used with ingestionType and ingest.Ingestor func createRealIngestor(client localClient, database string, tableName string) (localIngestor, error) { ingestor, err := ingest.New(client.(*kusto.Client), database, tableName, ingest.WithStaticBuffer(bufferSize, maxBuffers)) if ingestor != nil { diff --git a/plugins/outputs/azure_data_explorer/azure_data_explorer_test.go b/plugins/outputs/azure_data_explorer/azure_data_explorer_test.go index 7bb2b5c160add..83a1b13243f5a 100644 --- a/plugins/outputs/azure_data_explorer/azure_data_explorer_test.go +++ b/plugins/outputs/azure_data_explorer/azure_data_explorer_test.go @@ -255,9 +255,7 @@ func TestWriteWithType(t *testing.T) { }, } for _, testCase := range testCases { - testCase := testCase t.Run(testCase.name, func(t *testing.T) { - t.Parallel() serializer, err := telegrafJson.NewSerializer(time.Second, "", "") require.NoError(t, err) for tableName, jsonValue := range testCase.tableNameToExpectedResult {