elasticsearch indexing of result

docs/storing_results.md

@@ -113,7 +113,7 @@ Each service that is deployed gets an entry in the `jobResults.jobs` property. I
 
 Use `--sql [connection-string] --table [table-name]` arguments to store in the specified SQL Server database. The connection string must point to an existing SQL Server database. The first time it's called the required table will be created.
 
-The create table has the following schema:
+The created table has the following schema:
 
 
 | Column         | Type     | Example     | Description     |
@@ -125,3 +125,9 @@ The create table has the following schema:
 | Scenario | `nvarchard(200)` | hello | name of the scenario that was used |
 | Description | `nvarchard(200)` | custom string representing extra information about the job | |
 | Document | `nvarchar(max)` | { jobResults: {} } | json document containing the results of the job |
+
+### Elasticsearch
+
+Use `--es [server-url] --index [index-name]` arguments to store in the specified Elasticsearch server. The url must point to an existing Elasticsearch instance. The first time it's called the mapping of the required index will be created.
+
+The created index has the same set of fields as SqlServer

src/Microsoft.Crank.Controller/JobSerializer.cs

@@ -8,6 +8,9 @@
 using Microsoft.Crank.Models;
 using Newtonsoft.Json;
 using Newtonsoft.Json.Serialization;
+using System.Net.Http;
+using System.Text;
+using System.Net;
 
 namespace Microsoft.Crank.Controller.Serializers
 {
@@ -130,6 +133,88 @@ INSERT INTO [dbo].[" + tableName + @"]
                 }
             }
         }
+ public static Task WriteJobResultsToEsAsync(
+                 JobResults jobResults,
+                 string elasticSearchUrl,
+                 string indexName,
+                 string session,
+                 string scenario,
+                 string description
+                 )
+      {
+          var utcNow = DateTime.UtcNow;
+          return RetryOnExceptionAsync(5, () =>
+               WriteResultsToEs(
+                  utcNow,
+                  elasticSearchUrl,
+                  indexName,
+                  session,
+                  scenario,
+                  description,
+                  jobResults
+                  )
+              , 5000);
+      }
+
+      public static async Task InitializeElasticSearchAsync(string elasticSearchUrl, string indexName)
+      {
+          var mappingQuery =
+              @"{""settings"": {""number_of_shards"": 1},""mappings"": { ""dynamic"": false, 
+              ""properties"": {
+              ""DateTimeUtc"": { ""type"": ""date"" },
+              ""Session"": { ""type"": ""keyword"" },
+              ""Scenario"": { ""type"": ""keyword"" },
+              ""Description"": { ""type"": ""keyword"" },
+              ""Document"": { ""type"": ""nested"" }}}}";
+
+          await RetryOnExceptionAsync(5, () => InitializeDatabaseInternalAsync(elasticSearchUrl,indexName, mappingQuery), 5000);
+
+          static async Task InitializeDatabaseInternalAsync(string elasticSearchUrl,string indexName, string mappingQuery)
+          {
+              using (var httpClient = new HttpClient())
+              {
+                  HttpResponseMessage hrm = await httpClient.SendAsync(new HttpRequestMessage(HttpMethod.Head, $"{elasticSearchUrl}/{indexName.ToLower()}"));
+                  if (hrm.StatusCode == HttpStatusCode.NotFound)
+                  {
+                      hrm = await httpClient.PutAsync($"{elasticSearchUrl}/{indexName.ToLower()}", new StringContent(mappingQuery, Encoding.UTF8, "application/json"));
+                      if (hrm.StatusCode != HttpStatusCode.OK)
+                      {
+                          throw new System.Exception(await hrm.Content.ReadAsStringAsync());
+                      }
+                  }
+              }
+          }
+      }
+
+      private static async Task WriteResultsToEs(
+          DateTime utcNow,
+          string elasticSearchUrl,
+          string indexName,
+          string session,
+          string scenario,
+          string description,
+          JobResults jobResults
+          )
+      {
+          var result = new
+          {
+              DateTimeUtc = utcNow,
+              Session = session,
+              Scenario = scenario,
+              Description = description,
+              Document = jobResults
+          };
+
+          using (var httpClient = new HttpClient())
+          {
+              var item = JsonConvert.SerializeObject(result, Formatting.None, new JsonSerializerSettings { ContractResolver = new CamelCasePropertyNamesContractResolver() });
+              HttpResponseMessage hrm = await httpClient.PostAsync(elasticSearchUrl+"/"+indexName.ToLower() + "/_doc/" + session, new StringContent(item.ToString(), Encoding.UTF8, "application/json"));
+              if (!hrm.IsSuccessStatusCode)
+              {
+                  throw new Exception(hrm.RequestMessage?.ToString());
+              }
+          }
+      }
 
         private async static Task RetryOnExceptionAsync(int retries, Func<Task> operation, int milliSecondsDelay = 0)
         {