out_es: add cloud_apikey configuration

This patch adds support for the Elastic Cloud API Keys. It updates the
elastic search plugin to add a cloud_apikey configuration option. Usage:

[OUTPUT]
    name            es
    Cloud_Id        elastic_cloud_id
    Cloud_Apikey    elastic_apikey

Signed-off-by: Soedarsono <[email protected]>

Author: soedar

plugins/out_es/es.c

@@ -75,6 +75,27 @@ static flb_sds_t add_aws_auth(struct flb_http_client *c,
 }
 #endif /* FLB_HAVE_AWS */
 
+static int es_http_add_cloud_apikey(struct flb_http_client *c,
+                                    const char *apikey)
+{
+    flb_sds_t header;
+    int ret = 0;
+
+    if (apikey == NULL) {
+        return -1;
+    }
+
+    header = flb_sds_create_size(256 + 7);
+    header = flb_sds_printf(&header, "%s %s", "ApiKey", apikey);
+
+    ret = flb_http_add_header(c, "Authorization", 13,
+                                 header, flb_sds_len(header));
+
+    flb_sds_destroy(header);
+
+    return ret;
+}
+
 static int es_pack_map_content(msgpack_packer *tmp_pck,
                                msgpack_object map,
                                struct flb_elasticsearch *ctx)
@@ -875,11 +896,17 @@ static void cb_es_flush(struct flb_event_chunk *event_chunk,
     flb_http_add_header(c, "Content-Type", 12, "application/x-ndjson", 20);
 
     if (ctx->http_user && ctx->http_passwd) {
+        flb_plg_debug(ctx->ins, "using http basic auth");
         flb_http_basic_auth(c, ctx->http_user, ctx->http_passwd);
     }
     else if (ctx->cloud_user && ctx->cloud_passwd) {
+        flb_plg_debug(ctx->ins, "using elastic cloud auth");
         flb_http_basic_auth(c, ctx->cloud_user, ctx->cloud_passwd);
     }
+    else if (ctx->cloud_apikey) {
+        flb_plg_debug(ctx->ins, "using elastic cloud apikey");
+        es_http_add_cloud_apikey(c, ctx->cloud_apikey);
+    }
 
 #ifdef FLB_HAVE_AWS
     if (ctx->has_aws_auth == FLB_TRUE) {
@@ -937,7 +964,7 @@ static void cb_es_flush(struct flb_event_chunk *event_chunk,
                     /*
                      * If trace_error is set, trace the actual
                      * response from Elasticsearch explaining the problem.
-                     * Trace_Output can be used to see the request. 
+                     * Trace_Output can be used to see the request.
                      */
                     if (pack_size < 4000) {
                         flb_plg_debug(ctx->ins, "error caused by: Input\n%.*s\n",
@@ -1102,7 +1129,7 @@ static struct flb_config_map config_map[] = {
      "Set payload compression mechanism. Option available is 'gzip'"
     },
 
-    /* Cloud Authentication */
+    /* Elastic Cloud Authentication */
     {
      FLB_CONFIG_MAP_STR, "cloud_id", NULL,
      0, FLB_FALSE, 0,
@@ -1113,6 +1140,11 @@ static struct flb_config_map config_map[] = {
      0, FLB_FALSE, 0,
      "Elastic cloud authentication credentials"
     },
+    {
+     FLB_CONFIG_MAP_STR, "cloud_apikey", NULL,
+     0, FLB_FALSE, 0,
+     "Elastic cloud apikey credentials"
+    },
 
     /* AWS Authentication */
 #ifdef FLB_HAVE_AWS

plugins/out_es/es.h

@@ -59,6 +59,9 @@ struct flb_elasticsearch {
     char *cloud_user;
     char *cloud_passwd;
 
+    /* Elastic Cloud Apikey */
+    char *cloud_apikey;
+
     /* AWS Auth */
 #ifdef FLB_HAVE_AWS
     int has_aws_auth;

plugins/out_es/es_conf.c

@@ -224,6 +224,12 @@ struct flb_elasticsearch *flb_es_conf_create(struct flb_output_instance *ins,
         set_cloud_credentials(ctx, tmp);
     }
 
+   /* handle cloud_apikey */
+    tmp = flb_output_get_property("cloud_apikey", ins);
+    if (tmp) {
+        ctx->cloud_apikey = flb_strdup(tmp);
+    }
+
     /* use TLS ? */
     if (ins->use_tls == FLB_TRUE) {
         io_flags = FLB_IO_TLS;
@@ -531,6 +537,7 @@ int flb_es_conf_destroy(struct flb_elasticsearch *ctx)
 
     flb_free(ctx->cloud_passwd);
     flb_free(ctx->cloud_user);
+    flb_free(ctx->cloud_apikey);
     flb_free(ctx);
 
     return 0;