azure / webjobs.extensions.kusto Goto Github PK

I'm trying to use the Kusto output binding to write data into a Kusto table. However, I'm getting the following exception when trying to use the output binding.

Kusto.Ingest.Exceptions.NonexistentSourceIdException: Failed to query ingestion status for source with ID:'7b2bf49d-d667-44ba-b46a-1057e8392a80'. A source with the provided ID does not exist.
  HResult=0x80131500
  Source=Kusto.Ingest
  StackTrace:
   at Kusto.Ingest.KustoIngestionResult.GetIngestionStatusBySourceId(Guid sourceId)
   at Microsoft.Azure.WebJobs.Kusto.KustoAsyncCollector`1.<IngestData>d__9.MoveNext()
   at Microsoft.Azure.WebJobs.Kusto.KustoAsyncCollector`1.<IngestRowsAsync>d__8.MoveNext()
   at Microsoft.Azure.WebJobs.Kusto.KustoAsyncCollector`1.<FlushAsync>d__7.MoveNext()
   at Microsoft.Azure.WebJobs.Host.Bindings.OutArrayValueProvider`1.<SetValueAsync>d__8.MoveNext() in D:\a\_work\1\s\src\Microsoft.Azure.WebJobs.Host\Bindings\AsyncCollector\OutArrayValueProvider.cs:line 60

  This exception was originally thrown at this call stack:
    Kusto.Ingest.KustoIngestionResult.GetIngestionStatusBySourceId(System.Guid)
    Microsoft.Azure.WebJobs.Kusto.KustoAsyncCollector<T>.IngestData(string, Kusto.Ingest.KustoIngestionProperties, Kusto.Ingest.StreamSourceOptions)
    Microsoft.Azure.WebJobs.Kusto.KustoAsyncCollector<T>.IngestRowsAsync(System.Guid)
    Microsoft.Azure.WebJobs.Kusto.KustoAsyncCollector<T>.FlushAsync(System.Threading.CancellationToken)
    Microsoft.Azure.WebJobs.Host.Bindings.OutArrayValueProvider<TMessage>.SetValueAsync(object, System.Threading.CancellationToken) in OutArrayValueProvider.cs

I'm using the following package reference:

<PackageReference Include="Microsoft.Azure.WebJobs.Extensions.Kusto" Version="1.0.7-Preview" />

Here is my function code:

public record EmbeddingsRequest(string FilePath);

public record EmailEmbedding(string Subject, string Text, IReadOnlyList<double> Embeddings, DateTime Timestamp);

[FunctionName("EmailPrompt")]
public static IActionResult IngestEmail(
    [HttpTrigger(AuthorizationLevel.Function, "post")] EmbeddingsRequest req,
    [Embeddings("{FilePath}", InputType.FilePath, MaxChunkLength = 512)] EmbeddingCreateResponse embeddingsResponse,
    [Kusto("testing", TableName = "EmailEmbeddings", Connection = "KustoConnectionString")] out EmailEmbedding[] output,
    ILogger log)
{
    output = embeddingsResponse.Data.Select(
        d => new EmailEmbedding(
            Path.GetFileName(req.FilePath),
            "TODO: Text chunk",
            d.Embedding,
            DateTime.UtcNow)).ToArray();

    return new OkObjectResult(new
    {
        status = "success",
        count = output.Length,
    });
}

...and my connection string (copied from Kusto Explorer):

"KustoConnectionString": "Data Source=https://test-kusto-cluster.westus2.kusto.windows.net;Initial Catalog=NetDefaultDB;AAD Federated Security=True"

...which corresponds to an actual database in my Kusto cluster.

The error message isn't really helpful, and searching for it online doesn't provide any information. Any ideas what might be causing my problem?

problems converting Kusto Boolean data types.

Given this table structure, where the property IsHealthy is of type bool:

If I create an input binding with casting to my typed class:

With the Heartbeat record being:

I get this Exception:

As it turns out, the input binding cast logic seems to convert a boolean in to a Number:

(If I temporarily let it convert to an Enumerable of Objects instead):

That would explain the above Exception.

Now, my workaround is to assign a Custom Converter to that Boolean property:

Then it seems to work fine.

I wanted to use this extension along with my Managed Identity.
On local systems, this should default to the logged in user in Azure CLI or Visual Studio. However, this does not appear to be possible. I'm receiving the following error:

[2023-11-14T20:26:23.583Z] System.Private.CoreLib: Exception while executing function: Functions.AddSensorData. Microsoft.Azure.WebJobs.Host: Error while handling parameter _binder after function returned:. Kusto.Ingest: A permanent error occurred while attempting to ingest: 'Stream' via streaming ingestion. Error: 'ManagedIdentityCredential authentication unavailable. Multiple attempts failed to obtain a token from the managed identity endpoint.'. Azure.Identity: ManagedIdentityCredential authentication unavailable. Multiple attempts failed to obtain a token from the managed identity endpoint. Azure.Core: Retry failed after 4 tries. Retry settings can be adjusted in ClientOptions.Retry or by configuring a custom retry policy in ClientOptions.RetryPolicy. (A socket operation was attempted to an unreachable network. (169.254.169.254:80)) (A socket operation was attempted to an unreachable network. (169.254.169.254:80)) (A socket operation was attempted to an unreachable network. (169.254.169.254:80)) (A socket operation was attempted to an unreachable network. (169.254.169.254:80)). Azure.Core: A socket operation was attempted to an unreachable network. (169.254.169.254:80). System.Net.Http: A socket operation was attempted to an unreachable network. (169.254.169.254:80). System.Net.Sockets: A socket operation was attempted to an unreachable network.

As for context, this is the configuration I'm using

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
    "MachineDataDatabase": "data",
    "KustoConnectionString": "Data Source=https://my-instance.westeurope.kusto.windows.net;Initial Catalog=NetDefaultDB;User ID=;Password=;Application Client Id=;Application Key=;Application Certificate Thumbprint=;Application Certificate Subject Distinguished Name=;Application Certificate Issuer Distinguished Name=;Application Token=;User Token=;AAD Federated Security=True;dSTS Federated Security=False;Authority Id=",
    "MachineTemperatureTableName": "MachineTemperature"
  }
}

I got the connection string via LinqPad using the following code:

var kustoUri = "https://my-instance.westeurope.kusto.windows.net";
var kustoConnectionStringBuilder = new KustoConnectionStringBuilder(kustoUri)
	.WithAadSystemManagedIdentity();
	
kustoConnectionStringBuilder.ConnectionString.Dump();

The actual Azure Function is rather simple at the moment.

[Function(nameof(AddSensorData))]
[KustoOutput(Database: "data", Connection = "KustoConnectionString", TableName = "MachineTemperature", ManagedServiceIdentity = "system")]
public IEnumerable<MachineTemperature> AddSensorData(
    [HttpTrigger(AuthorizationLevel.Function, "get")]
    HttpRequestData req)
{
    _logger.LogInformation("Creating temperature data.");

    List<MachineTemperature> result = new ();
    for(int i = 0; i < Produce.BatchSize; i++)
    {
        var machineName = NameBuilder.Machine(random.Next(0, Produce.BatchSize));
        result.Add(new MachineTemperature
        {
            MachineId = computeIdentifier.Invoke(machineName),
            MachineName = machineName,
            TimeGenerated = DateTime.UtcNow,
            TemperatureCelcius = random.Next(2000, 15000)
        });
    }
    return result;
}

Am I doing something wrong in my current setup, maybe the connectionstring? Or is this scenario not supported (yet)?
Would love to know if/how I can resolve this.

Of course, creating a service principal & secret is a possibility but that would be my last resort.

There is support for all text based formats in Output bindings. Adding support for Avro based bindings for ingestion will add additional capabilities in case of scenarios such as ingesting from Kafka

Line 14 of https://github.com/Azure/Webjobs.Extensions.Kusto/blob/main/samples/samples-outofproc/InputBindingSamples/GetProducts.cs should be [FunctionName("GetProducts")] instead of [Function("GetProducts")]