Add support for scripted hash updates (#117)

* Add support for scripted hash updates

* Include a hash of the channel options

Author: Mpdreamz

src/Elastic.Ingest.Elasticsearch/Catalog/CatalogIndexChannel.cs

@@ -28,11 +28,40 @@ public class CatalogIndexChannelOptionsBase<TDocument>(ITransport transport) : I
 /// <inheritdoc cref="CatalogIndexChannel{TDocument}" />
 public class CatalogIndexChannelOptions<TDocument>(ITransport transport) : CatalogIndexChannelOptionsBase<TDocument>(transport)
 {
+	private readonly Func<string>? _getMapping;
+
 	/// A function that returns the mapping for <typeparamref name="TDocument"/>.
-	public Func<string>? GetMapping { get; init; }
+	public Func<string>? GetMapping
+	{
+		get => _getMapping;
+		init
+		{
+			_getMapping = value;
+			_channelHash = HashedBulkUpdate.CreateHash(base.ChannelHash, ActiveSearchAlias,
+				_getMapping?.Invoke() ?? string.Empty, _getMapping?.Invoke() ?? string.Empty
+			);
+		}
+	}
+
+	private readonly Func<string>? _getMappingSettings;
 
 	/// A function that returns settings to accompany <see cref="GetMapping"/>.
-	public Func<string>? GetMappingSettings { get; init; }
+	public Func<string>? GetMappingSettings
+	{
+		get => _getMappingSettings;
+		init
+		{
+			_getMappingSettings = value;
+			_channelHash = HashedBulkUpdate.CreateHash(base.ChannelHash, ActiveSearchAlias,
+				_getMapping?.Invoke() ?? string.Empty, _getMapping?.Invoke() ?? string.Empty
+			);
+		}
+	}
+
+	private readonly string _channelHash = string.Empty;
+
+	/// <inheritdoc />
+	public override string ChannelHash => string.IsNullOrEmpty(_channelHash) ? base.ChannelHash : _channelHash;
 }
 
 /// <inheritdoc cref="CatalogIndexChannel{TDocument}" />
@@ -61,7 +90,7 @@ public class CatalogIndexChannel<TDocument, TChannelOptions> : IndexChannel<TDoc
 	where TChannelOptions : CatalogIndexChannelOptionsBase<TDocument>
 	where TDocument : class
 {
-	private readonly string _url;
+	private string _url;
 
 	/// <inheritdoc cref="CatalogIndexChannel{TDocument}"/>
 	public CatalogIndexChannel(TChannelOptions options, ICollection<IChannelCallbacks<TDocument, BulkResponse>>? callbackListeners = null)
@@ -82,7 +111,7 @@ public CatalogIndexChannel(TChannelOptions options, ICollection<IChannelCallback
 
 	/// The index name used for indexing. This the configured <see cref="IndexChannelOptions{TDocument}.IndexFormat"/> to compute a single index name for all operations.
 	/// <para>Since this is catalog data and not time series data, all data needs to end up in a single index</para>
-	public string IndexName { get; }
+	public string IndexName { get; private set; }
 
 	/// <inheritdoc cref="ElasticsearchChannelBase{TEvent,TChannelOptions}.CreateBulkOperationHeader"/>
 	protected override BulkOperationHeader CreateBulkOperationHeader(TDocument @event) =>
@@ -97,6 +126,38 @@ protected override BulkOperationHeader CreateBulkOperationHeader(TDocument @even
 	/// <inheritdoc cref="ElasticsearchChannelBase{TEvent,TChannelOptions}.AlwaysBootstrapComponentTemplates"/>
 	protected override bool AlwaysBootstrapComponentTemplates => true;
 
+	/// <inheritdoc />
+	public override async Task<bool> BootstrapElasticsearchAsync(BootstrapMethod bootstrapMethod, string? ilmPolicy = null, CancellationToken ctx = default)
+	{
+		if (Options.ScriptedHashBulkUpsertLookup is null)
+			return await base.BootstrapElasticsearchAsync(bootstrapMethod, ilmPolicy, ctx).ConfigureAwait(false);
+		var latestAlias = string.Format(Options.IndexFormat, "latest");
+		var matchingIndices = string.Format(Options.IndexFormat, "*");
+		var currentIndex = await ShouldRemovePreviousAliasAsync(matchingIndices, latestAlias, ctx).ConfigureAwait(false);
+		if (string.IsNullOrEmpty(currentIndex))
+			return await base.BootstrapElasticsearchAsync(bootstrapMethod, ilmPolicy, ctx).ConfigureAwait(false);
+
+		IndexName = currentIndex;
+		_url = $"{IndexName}/{base.BulkPathAndQuery}";
+		return await base.BootstrapElasticsearchAsync(bootstrapMethod, ilmPolicy, ctx).ConfigureAwait(false);
+	}
+
+	/// <inheritdoc />
+	public override bool BootstrapElasticsearch(BootstrapMethod bootstrapMethod, string? ilmPolicy = null)
+	{
+		if (Options.ScriptedHashBulkUpsertLookup is null)
+			return base.BootstrapElasticsearch(bootstrapMethod, ilmPolicy);
+		var latestAlias = string.Format(Options.IndexFormat, "latest");
+		var matchingIndices = string.Format(Options.IndexFormat, "*");
+		var currentIndex = ShouldRemovePreviousAlias(matchingIndices, latestAlias);
+		if (string.IsNullOrEmpty(currentIndex))
+			return base.BootstrapElasticsearch(bootstrapMethod, ilmPolicy);
+
+		IndexName = currentIndex;
+		_url = $"{IndexName}/{base.BulkPathAndQuery}";
+		return base.BootstrapElasticsearch(bootstrapMethod, ilmPolicy);
+	}
+
 	/// Applies the latest alias to the index.
 	public async Task<bool> ApplyLatestAliasAsync(CancellationToken ctx = default)
 	{
@@ -192,6 +253,12 @@ private async Task<string> ShouldRemovePreviousAliasAsync(string matchingIndices
 			.Trim(Environment.NewLine.ToCharArray());
 		return hasPreviousVersions.ApiCallDetails.HttpStatusCode == 200 ? queryAliasIndex : string.Empty;
 	}
+	private string ShouldRemovePreviousAlias(string matchingIndices, string alias)
+	{
+		var hasPreviousVersions = Options.Transport.Head($"{matchingIndices}?allow_no_indices=false");
+		var queryAliasIndex = Cat($"_cat/aliases/{alias}?h=index").Trim(Environment.NewLine.ToCharArray());
+		return hasPreviousVersions.ApiCallDetails.HttpStatusCode == 200 ? queryAliasIndex : string.Empty;
+	}
 
 	private async Task<string> CatAsync(string url, CancellationToken ctx)
 	{
@@ -201,6 +268,13 @@ private async Task<string> CatAsync(string url, CancellationToken ctx)
 		return catResponse.Body;
 	}
 
+	private string Cat(string url)
+	{
+		var rq = new RequestConfiguration { Accept = "text/plain" };
+		var catResponse = Options.Transport.Request<StringResponse>(new EndpointPath(HttpMethod.GET, url), postData: null, null, rq);
+		return catResponse.Body;
+	}
+
 
 	/// Applies the latest and current aliases to <see cref="IndexName"/>. Use this if you want to ensure that the latest index is always the active index
 	/// immediately after writing to the index.

src/Elastic.Ingest.Elasticsearch/ElasticsearchChannelBase.cs

@@ -53,7 +53,7 @@ protected override bool Retry(BulkResponse response)
 	/// <summary>
 	/// The URL for the bulk request.
 	/// </summary>
-	protected virtual string BulkPathAndQuery => "_bulk?filter_path=error,items.*.status,items.*.error";
+	protected virtual string BulkPathAndQuery => "_bulk?filter_path=error,items.*.status,items.*.error,items.*.result,items.*._version";
 
 	/// <inheritdoc cref="ResponseItemsBufferedChannelBase{TChannelOptions,TEvent,TResponse,TBulkResponseItem}.RetryAllItems"/>
 	protected override bool RetryAllItems(BulkResponse response) => response.ApiCallDetails.HttpStatusCode == 429;

src/Elastic.Ingest.Elasticsearch/ElasticsearchChannelStatics.cs

@@ -17,6 +17,17 @@ internal static class ElasticsearchChannelStatics
 	public static readonly byte[] DocUpdateHeaderStart = "{\"doc_as_upsert\": true, \"doc\": "u8.ToArray();
 	public static readonly byte[] DocUpdateHeaderEnd = " }"u8.ToArray();
 
+	public static readonly byte[] ScriptedHashUpsertStart =
+		"{ \"scripted_upsert\": true, \"upsert\": {}, \"script\": { \"source\": \"if (ctx.op != 'create') { if (ctx._source."u8.ToArray();
+
+	public static readonly byte[] ScriptedHashUpsertMiddle =
+		" == params.hash ) { ctx.op = 'noop' } } ctx._source = params.doc\", \"params\": { \"hash\": "u8.ToArray();
+
+	public static readonly byte[] ScriptedHashUpsertDocPreamble =
+		", \"doc\":"u8.ToArray();
+
+	public static readonly byte[] ScriptedHashUpsertEnd = " } } }"u8.ToArray();
+
 	public static readonly HashSet<int> RetryStatusCodes = [502, 503, 504, 429];
 
 	public static readonly JsonSerializerOptions SerializerOptions = new ()

src/Elastic.Ingest.Elasticsearch/Indices/IndexChannelOptions.cs

@@ -2,10 +2,36 @@
 // Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
 // See the LICENSE file in the project root for more information
 using System;
+using System.Security.Cryptography;
+using System.Text;
 using Elastic.Transport;
 
 namespace Elastic.Ingest.Elasticsearch.Indices;
 
+/// <summary>
+/// For scripted hash bulk upserts returns per operation the field and the hash to use for the scripted upsert
+/// </summary>
+/// <param name="Field">The field to check the previous hash against</param>
+/// <param name="Hash">The current hash of the document</param>
+public record HashedBulkUpdate(string Field, string Hash)
+{
+	/// <summary>
+	/// A short SHA256 hash of the provided <paramref name="components"/>
+	/// </summary>
+	/// <param name="components"></param>
+	/// <returns></returns>
+	public static string CreateHash(params string[] components)
+	{
+#if NET8_0_OR_GREATER
+		return Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(string.Join("", components))))[..8];
+#else
+		var sha = SHA256.Create();
+		var hash = sha.ComputeHash(Encoding.UTF8.GetBytes(string.Join("", components)));
+		return BitConverter.ToString(hash).Replace("-", "")[..8];
+#endif
+	}
+};
+
 /// <summary>
 /// Provides options to <see cref="IndexChannel{TEvent}"/> to control how and where data gets written to Elasticsearch
 /// </summary>
@@ -15,27 +41,47 @@ public class IndexChannelOptions<TEvent> : ElasticsearchChannelOptionsBase<TEven
 	/// <inheritdoc cref="IndexChannelOptions{TEvent}"/>
 	public IndexChannelOptions(ITransport transport) : base(transport) { }
 
+	private string _indexFormat = $"{typeof(TEvent).Name.ToLowerInvariant()}-{{0:yyyy.MM.dd}}";
+
 	/// <summary>
 	/// Gets or sets the format string for the Elastic search index. The current <c>DateTimeOffset</c> is passed as parameter
 	/// 0.
 	/// <para> Defaults to "<typeparamref name="TEvent"/>.Name.ToLowerInvariant()-{0:yyyy.MM.dd}"</para>
 	/// <para> If no {0} parameter is defined the index name is effectively fixed</para>
 	/// </summary>
-	public virtual string IndexFormat { get; set; } = $"{typeof(TEvent).Name.ToLowerInvariant()}-{{0:yyyy.MM.dd}}";
+	public virtual string IndexFormat
+	{
+		get => _indexFormat;
+		set
+		{
+			_indexFormat = value;
+			_channelHash = HashedBulkUpdate.CreateHash(_indexFormat, _indexOffset?.ToString() ?? string.Empty);
+		}
+	}
+
+	private TimeSpan? _indexOffset;
 
 	/// <summary>
 	/// Gets or sets the offset to use for the index <c>DateTimeOffset</c>. The default value is null, which uses the system local
 	/// offset. Use "00:00" for UTC.
 	/// </summary>
-	public TimeSpan? IndexOffset { get; set; }
+	public TimeSpan? IndexOffset
+	{
+		get => _indexOffset;
+		set
+		{
+			_indexOffset = value;
+			_channelHash = HashedBulkUpdate.CreateHash(_indexFormat, _indexOffset?.ToString() ?? string.Empty);
+		}
+	}
 
 	/// <summary>
 	/// Provide a per document <c>DateTimeOffset</c> to be used as the date passed as parameter 0 to <see cref="IndexFormat"/>
 	/// </summary>
 	public Func<TEvent, DateTimeOffset?>? TimestampLookup { get; set; }
 
 	/// <summary>
-	/// If the document provides an I, D, this allows you to set a per document `_id`.
+	/// If the document provides an ID, this allows you to set a per document `_id`.
 	/// <para>If an `_id` is defined, an `_index` bulk operation will be created.</para>
 	/// <para>Otherwise (the default) `_create` bulk operation will be issued for the document.</para>
 	/// <para>Read more about bulk operations here:</para>
@@ -52,8 +98,26 @@ public IndexChannelOptions(ITransport transport) : base(transport) { }
 	/// </summary>
 	public Func<TEvent, string, bool>? BulkUpsertLookup { get; set; }
 
+	/// <summary>
+	/// Uses the callback provided to <see cref="BulkOperationIdLookup"/> to determine if this is in fact an update operation
+	/// <para>Returns the field and the hash to use for the scripted upsert </para>
+	/// <para>The string passed to the callback is the hash of the current channel options that can be used to hash bust in case of channel option changes</para>
+	/// <para>Otherwise (the default) `index` bulk operation will be issued for the document.</para>
+	/// <para>Read more about bulk operations here:</para>
+	/// <para>https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html#bulk-api-request-body</para>
+	/// </summary>
+	public Func<TEvent, string, HashedBulkUpdate>? ScriptedHashBulkUpsertLookup { get; set; }
+
 	/// <summary>
 	/// Control the operation header for each bulk operation.
 	/// </summary>
 	public OperationMode OperationMode { get; set; }
+
+	private string _channelHash = HashedBulkUpdate.CreateHash($"{typeof(TEvent).Name.ToLowerInvariant()}-{{0:yyyy.MM.dd}}");
+	/// <summary>
+	/// A hash of the channel options. This can be used to hash bust in case of channel option changes.
+	/// It's also passed to <see cref="ScriptedHashBulkUpsertLookup"/>
+	/// </summary>
+	public virtual string ChannelHash => _channelHash;
+
 }

src/Elastic.Ingest.Elasticsearch/Serialization/BulkOperationHeader.cs

@@ -5,11 +5,12 @@
 using System.Collections.Generic;
 using System.Text.Json;
 using System.Text.Json.Serialization;
+using Elastic.Ingest.Elasticsearch.Indices;
 using Elastic.Transport.Products.Elasticsearch;
 
 namespace Elastic.Ingest.Elasticsearch.Serialization;
 
-/// <summary> Represents the _bulk operation meta header </summary>
+/// <summary> Represents the _bulk operation meta-header </summary>
 public abstract class BulkOperationHeader
 {
 	/// <summary> The index or data stream to write to </summary>
@@ -25,7 +26,7 @@ public abstract class BulkOperationHeader
 	public bool? RequireAlias { get; init; }
 }
 
-/// <summary> Represents the _bulk create operation meta header </summary>
+/// <summary> Represents the _bulk create operation meta-header </summary>
 [JsonConverter(typeof(BulkOperationHeaderConverter<CreateOperation>))]
 public class CreateOperation : BulkOperationHeader
 {
@@ -34,7 +35,7 @@ public class CreateOperation : BulkOperationHeader
 	public Dictionary<string, string>? DynamicTemplates { get; init; }
 }
 
-/// <summary> Represents the _bulk index operation meta header </summary>
+/// <summary> Represents the _bulk index operation meta-header </summary>
 [JsonConverter(typeof(BulkOperationHeaderConverter<IndexOperation>))]
 public class IndexOperation : BulkOperationHeader
 {
@@ -43,16 +44,20 @@ public class IndexOperation : BulkOperationHeader
 	public Dictionary<string, string>? DynamicTemplates { get; init; }
 }
 
-/// <summary> Represents the _bulk delete operation meta header </summary>
+/// <summary> Represents the _bulk delete operation meta-header </summary>
 [JsonConverter(typeof(BulkOperationHeaderConverter<DeleteOperation>))]
-public class DeleteOperation : BulkOperationHeader
-{
-}
+public class DeleteOperation : BulkOperationHeader;
 
-/// <summary> Represents the _bulk update operation meta header </summary>
+/// <summary> Represents the _bulk update operation meta-header </summary>
 [JsonConverter(typeof(BulkOperationHeaderConverter<UpdateOperation>))]
-public class UpdateOperation : BulkOperationHeader
+public class UpdateOperation : BulkOperationHeader;
+
+/// <summary> Represents the _bulk update operation meta-header </summary>
+[JsonConverter(typeof(BulkOperationHeaderConverter<ScriptedHashUpdateOperation>))]
+public class ScriptedHashUpdateOperation : BulkOperationHeader
 {
+	/// <summary>  </summary>
+	public required HashedBulkUpdate UpdateInformation { get; init; }
 }
 
 internal class BulkOperationHeaderConverter<THeader> : JsonConverter<THeader>
@@ -69,6 +74,7 @@ public override void Write(Utf8JsonWriter writer, THeader value, JsonSerializerO
 			DeleteOperation _ => "delete",
 			IndexOperation _ => "index",
 			UpdateOperation _ => "update",
+			ScriptedHashUpdateOperation _ => "update",
 			_ => throw new ArgumentOutOfRangeException(nameof(value), value, null)
 		};
 		writer.WriteStartObject();