pipeline: outputs: es: basic support of Upstream, only parameters known to parser of Upstream node configuration section are implemented, e.g. "host" and "port"

Signed-off-by: Marat Abrarov <[email protected]>

Author: mabrarov

plugins/out_es/CMakeLists.txt

@@ -1,5 +1,6 @@
 set(src
   es_bulk.c
+  es_conf_parse.c
   es_conf.c
   es.c
   murmur3.c)

plugins/out_es/es.c

@@ -20,8 +20,11 @@
 #ifndef FLB_OUT_ES_H
 #define FLB_OUT_ES_H
 
+#include <monkey/mk_core/mk_list.h>
+#include <fluent-bit/flb_sds.h>
+
 #define FLB_ES_DEFAULT_HOST       "127.0.0.1"
-#define FLB_ES_DEFAULT_PORT       92000
+#define FLB_ES_DEFAULT_PORT       9200
 #define FLB_ES_DEFAULT_INDEX      "fluent-bit"
 #define FLB_ES_DEFAULT_TYPE       "_doc"
 #define FLB_ES_DEFAULT_PREFIX     "logstash"
@@ -36,10 +39,12 @@
 #define FLB_ES_WRITE_OP_UPDATE    "update"
 #define FLB_ES_WRITE_OP_UPSERT    "upsert"
 
-struct flb_elasticsearch {
+struct flb_elasticsearch_config {
     /* Elasticsearch index (database) and type (table) */
     char *index;
+    int own_index;
     char *type;
+    int own_type;
     char suppress_type_name;
 
     /* HTTP Auth */
@@ -60,9 +65,7 @@ struct flb_elasticsearch {
     struct flb_aws_provider *base_aws_provider;
     /* tls instances can't be re-used; aws provider requires a separate one */
     struct flb_tls *aws_tls;
-    /* one for the standard chain provider, one for sts assume role */
     struct flb_tls *aws_sts_tls;
-    char *aws_session_name;
     char *aws_service_name;
     struct mk_list *aws_unsigned_headers;
 #endif
@@ -108,7 +111,6 @@ struct flb_elasticsearch {
     /* time key nanoseconds */
     int time_key_nanos;
 
-
     /* write operation */
     flb_sds_t write_operation;
     /* write operation elasticsearch operation */
@@ -130,8 +132,19 @@ struct flb_elasticsearch {
     /* Compression mode (gzip) */
     int compress_gzip;
 
-    /* Upstream connection to the backend server */
+    /* List entry data for flb_elasticsearch->configs list */
+    struct mk_list _head;
+};
+
+struct flb_elasticsearch {
+    /* if HA mode is enabled */
+    int ha_mode;              /* High Availability mode enabled ? */
+    char *ha_upstream;        /* Upstream configuration file      */
+    struct flb_upstream_ha *ha;
+
+    /* Upstream handler and config context for single mode (no HA) */
     struct flb_upstream *u;
+    struct mk_list configs;
 
     /* Plugin output instance reference */
     struct flb_output_instance *ins;