Update Kusto Go SDK version , fixes issue open-telemetry#22771 (open-…

…telemetry#22933)

The underlying SDK has an update to fix an issue where the network connections are flaky. To get the metadata to connect to the service , a HTTP metadata call is issued. If this call fails , it is not retried and the init is interrupted till restart. The underlying SDK fixes it by reset of the lock that was causing this behavior

.chloggen/kusto_init_retry.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: azuredataexplorerexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Update underlying SDK to perform retries on init on machines with flaky networks.
+
+# One or more tracking issues related to the change
+issues: [22771]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

cmd/configschema/go.mod

@@ -229,7 +229,7 @@ require (
 	github.com/AthenZ/athenz v1.10.39 // indirect
 	github.com/Azure/azure-amqp-common-go/v4 v4.2.0 // indirect
 	github.com/Azure/azure-event-hubs-go/v3 v3.6.0 // indirect
-	github.com/Azure/azure-kusto-go v0.11.3 // indirect
+	github.com/Azure/azure-kusto-go v0.13.1 // indirect
 	github.com/Azure/azure-pipeline-go v0.2.3 // indirect
 	github.com/Azure/azure-sdk-for-go v68.0.0+incompatible // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/azcore v1.4.0 // indirect

cmd/otelcontribcol/go.mod

@@ -221,7 +221,7 @@ require (
 	github.com/AthenZ/athenz v1.10.39 // indirect
 	github.com/Azure/azure-amqp-common-go/v4 v4.2.0 // indirect
 	github.com/Azure/azure-event-hubs-go/v3 v3.6.0 // indirect
-	github.com/Azure/azure-kusto-go v0.11.3 // indirect
+	github.com/Azure/azure-kusto-go v0.13.1 // indirect
 	github.com/Azure/azure-pipeline-go v0.2.3 // indirect
 	github.com/Azure/azure-sdk-for-go v68.0.0+incompatible // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/azcore v1.3.1 // indirect

exporter/azuredataexplorerexporter/e2e_test.go

@@ -13,7 +13,7 @@ import (
 	"github.com/Azure/azure-kusto-go/kusto"
 	"github.com/Azure/azure-kusto-go/kusto/data/errors"
 	"github.com/Azure/azure-kusto-go/kusto/data/table"
-	"github.com/Azure/azure-kusto-go/kusto/data/types"
+	"github.com/Azure/azure-kusto-go/kusto/kql"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -69,12 +69,11 @@ func TestCreateTracesExporterE2E(t *testing.T) {
 	td, tID, attrs := createTraces()
 	err = exp.ConsumeTraces(context.Background(), td)
 	require.NoError(t, err)
-	kustoDefinitions, kustoParameters := prepareQuery(tID)
 	// Statements
-	traceStmt := kusto.NewStmt(traceValidationQuery)
-	traceStmt = traceStmt.MustDefinitions(kusto.NewDefinitions().Must(kustoDefinitions)).MustParameters(kusto.NewParameters().Must(kustoParameters))
+	traceStmt := kql.New(traceValidationQuery)
+	traceStmtParams := kql.NewParameters().AddString("TID", tID)
 	// Query using our trace table for TraceID
-	iter, err := createClientAndExecuteQuery(t, *config, traceStmt)
+	iter, err := createClientAndExecuteQuery(t, *config, traceStmt, traceStmtParams)
 	if err != nil {
 		assert.Fail(t, err.Error())
 	}
@@ -132,12 +131,10 @@ func TestCreateLogsExporterE2E(t *testing.T) {
 	ld, tID, attrs := createLogs()
 	err = exp.ConsumeLogs(context.Background(), ld)
 	require.NoError(t, err)
-	kustoDefinitions, kustoParameters := prepareQuery(tID)
 	// Statements
-	traceStmt := kusto.NewStmt(logValidationQuery)
-	traceStmt = traceStmt.MustDefinitions(kusto.NewDefinitions().Must(kustoDefinitions)).MustParameters(kusto.NewParameters().Must(kustoParameters))
-	// Query using our logs table for TraceID
-	iter, err := createClientAndExecuteQuery(t, *config, traceStmt)
+	traceStmt := kql.New(traceValidationQuery)
+	traceStmtParams := kql.NewParameters().AddString("TID", tID)
+	iter, err := createClientAndExecuteQuery(t, *config, traceStmt, traceStmtParams)
 	if err != nil {
 		assert.Fail(t, err.Error())
 	}
@@ -195,12 +192,11 @@ func TestCreateMetricsExporterE2E(t *testing.T) {
 	md, attrs, metricName := createMetrics()
 	err = exp.ConsumeMetrics(context.Background(), md)
 	require.NoError(t, err)
-	kustoDefinitions, kustoParameters := prepareQuery(metricName)
 	// Statements
-	traceStmt := kusto.NewStmt(metricValidationQuery)
-	traceStmt = traceStmt.MustDefinitions(kusto.NewDefinitions().Must(kustoDefinitions)).MustParameters(kusto.NewParameters().Must(kustoParameters))
+	traceStmt := kql.New(traceValidationQuery)
+	traceStmtParams := kql.NewParameters().AddString("TID", metricName)
 	// Query using our logs table for TraceID
-	iter, err := createClientAndExecuteQuery(t, *config, traceStmt)
+	iter, err := createClientAndExecuteQuery(t, *config, traceStmt, traceStmtParams)
 	if err != nil {
 		assert.Fail(t, err.Error())
 	}
@@ -235,14 +231,6 @@ func TestCreateMetricsExporterE2E(t *testing.T) {
 	t.Cleanup(func() { _ = exp.Shutdown(context.Background()) })
 }
 
-func prepareQuery(tID string) (kusto.ParamTypes, kusto.QueryValues) {
-	kustoDefinitions := make(kusto.ParamTypes)
-	kustoParameters := make(kusto.QueryValues)
-	kustoDefinitions["TID"] = kusto.ParamType{Type: types.String}
-	kustoParameters["TID"] = tID
-	return kustoDefinitions, kustoParameters
-}
-
 func getConfig() (*Config, bool) {
 	if os.Getenv(clusterURI) == "" || os.Getenv(appID) == "" || os.Getenv(appKey) == "" || os.Getenv(tenantID) == "" || os.Getenv(otelE2EDb) == "" {
 		return nil, false
@@ -336,7 +324,7 @@ func createMetrics() (pmetric.Metrics, map[string]interface{}, string) {
 	return tm, attrs, metricNameGUID
 }
 
-func createClientAndExecuteQuery(t *testing.T, config Config, stmt kusto.Stmt) (*kusto.RowIterator, error) {
+func createClientAndExecuteQuery(t *testing.T, config Config, query *kql.Builder, params *kql.Parameters) (*kusto.RowIterator, error) {
 	kcsb := kusto.NewConnectionStringBuilder(config.ClusterURI).WithAadAppKey(config.ApplicationID, string(config.ApplicationKey), config.TenantID)
 	client, kerr := kusto.New(kcsb)
 	// The client should be created
@@ -348,6 +336,7 @@ func createClientAndExecuteQuery(t *testing.T, config Config, stmt kusto.Stmt) (
 	return client.Query(
 		context.Background(),
 		config.Database,
-		stmt,
+		query,
+		kusto.QueryParameters(params),
 	)
 }

exporter/azuredataexplorerexporter/go.mod

@@ -3,7 +3,7 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/exporter/azured
 go 1.19
 
 require (
-	github.com/Azure/azure-kusto-go v0.11.3
+	github.com/Azure/azure-kusto-go v0.13.1
 	github.com/google/uuid v1.3.0
 	github.com/json-iterator/go v1.1.12
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.78.0
@@ -49,6 +49,7 @@ require (
 	github.com/pkg/browser v0.0.0-20210911075715-681adbf594b8 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/samber/lo v1.37.0 // indirect
+	github.com/shopspring/decimal v1.3.1 // indirect
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/collector/consumer v0.78.3-0.20230525165144-87dd85a6c034 // indirect
 	go.opentelemetry.io/collector/featuregate v1.0.0-rcv0012.0.20230525165144-87dd85a6c034 // indirect

go.mod

@@ -201,7 +201,7 @@ require (
 	github.com/AthenZ/athenz v1.10.39 // indirect
 	github.com/Azure/azure-amqp-common-go/v4 v4.2.0 // indirect
 	github.com/Azure/azure-event-hubs-go/v3 v3.6.0 // indirect
-	github.com/Azure/azure-kusto-go v0.11.3 // indirect
+	github.com/Azure/azure-kusto-go v0.13.1 // indirect
 	github.com/Azure/azure-pipeline-go v0.2.3 // indirect
 	github.com/Azure/azure-sdk-for-go v68.0.0+incompatible // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/azcore v1.4.0 // indirect