out_s3: Handle otlp JSON format

Signed-off-by: Hiroshi Hatake <hiroshi@chronosphere.io>

Author: cosmo0920

plugins/out_s3/s3.c

@@ -22,6 +22,7 @@
 #include <fluent-bit/flb_slist.h>
 #include <fluent-bit/flb_time.h>
 #include <fluent-bit/flb_pack.h>
+#include <fluent-bit/flb_opentelemetry.h>
 #include <fluent-bit/flb_config_map.h>
 #include <fluent-bit/flb_aws_util.h>
 #include <fluent-bit/aws/flb_aws_compress.h>
@@ -84,6 +85,10 @@ static void remove_from_queue(struct upload_queue *entry);
 
 static int blob_initialize_authorization_endpoint_upstream(struct flb_s3 *context);
 
+static flb_sds_t s3_format_event_chunk(struct flb_s3 *ctx,
+                                       struct flb_event_chunk *event_chunk,
+                                       struct flb_config *config);
+
 static struct flb_aws_header *get_content_encoding_header(int compression_type)
 {
     static struct flb_aws_header gzip_header = {
@@ -649,6 +654,7 @@ static int cb_s3_init(struct flb_output_instance *ins,
 
     ctx->retry_time = 0;
     ctx->upload_queue_success = FLB_FALSE;
+    ctx->out_format = FLB_PACK_JSON_FORMAT_LINES;
 
     /*
      * The engine default retry_limit (1) is too low for S3's internal
@@ -684,6 +690,23 @@ static int cb_s3_init(struct flb_output_instance *ins,
                      "S3 has its own buffer files located in the store_dir.");
     }
 
+    /* Format key */
+    tmp = flb_output_get_property("format", ins);
+    if (tmp) {
+        ret = flb_pack_to_json_format_type(tmp);
+        if (ret == -1) {
+            flb_plg_error(ctx->ins, "invalid format '%s'", tmp);
+            return -1;
+        }
+
+        if (ret != FLB_PACK_JSON_FORMAT_LINES &&
+            ret != FLB_PACK_JSON_FORMAT_OTLP) {
+            flb_plg_error(ctx->ins, "unsupported format '%s'", tmp);
+            return -1;
+        }
+        ctx->out_format = ret;
+    }
+
     /* Date key */
     ctx->date_key = ctx->json_date_key;
     tmp = flb_output_get_property("json_date_key", ins);
@@ -3765,6 +3788,58 @@ static int s3_timer_create(struct flb_s3 *ctx)
     return 0;
 }
 
+static flb_sds_t s3_format_event_chunk(struct flb_s3 *ctx,
+                                       struct flb_event_chunk *event_chunk,
+                                       struct flb_config *config)
+{
+    int result;
+    static const char *default_logs_body_keys[] = {"log", "message"};
+    struct flb_opentelemetry_otlp_logs_options options;
+
+    if (ctx->out_format == FLB_PACK_JSON_FORMAT_OTLP) {
+        if (event_chunk->type == FLB_EVENT_TYPE_LOGS) {
+            memset(&options, 0, sizeof(options));
+            options.logs_require_otel_metadata = FLB_FALSE;
+            options.logs_body_keys = default_logs_body_keys;
+            options.logs_body_key_count = 2;
+            options.logs_body_key_attributes = FLB_FALSE;
+
+            return flb_opentelemetry_logs_to_otlp_json(event_chunk->data,
+                                                       event_chunk->size,
+                                                       &options,
+                                                       &result);
+        }
+#ifdef FLB_HAVE_METRICS
+        else if (event_chunk->type == FLB_EVENT_TYPE_METRICS) {
+            return flb_opentelemetry_metrics_msgpack_to_otlp_json(event_chunk->data,
+                                                                  event_chunk->size,
+                                                                  &result);
+        }
+#endif
+        else if (event_chunk->type == FLB_EVENT_TYPE_TRACES) {
+            return flb_opentelemetry_traces_msgpack_to_otlp_json(event_chunk->data,
+                                                                 event_chunk->size,
+                                                                 &result);
+        }
+
+        return NULL;
+    }
+
+    if (ctx->log_key) {
+        return flb_pack_msgpack_extract_log_key(ctx,
+                                                event_chunk->data,
+                                                event_chunk->size,
+                                                config);
+    }
+
+    return flb_pack_msgpack_to_json_format(event_chunk->data,
+                                           event_chunk->size,
+                                           FLB_PACK_JSON_FORMAT_LINES,
+                                           ctx->json_date_format,
+                                           ctx->date_key,
+                                           config->json_escape_unicode);
+}
+
 static void cb_s3_flush(struct flb_event_chunk *event_chunk,
                         struct flb_output_flush *out_flush,
                         struct flb_input_instance *i_ins,
@@ -3800,20 +3875,7 @@ static void cb_s3_flush(struct flb_event_chunk *event_chunk,
     flush_init(ctx);
 
     /* Process chunk */
-    if (ctx->log_key) {
-        chunk = flb_pack_msgpack_extract_log_key(ctx,
-                                                 event_chunk->data,
-                                                 event_chunk->size,
-                                                 config);
-    }
-    else {
-        chunk = flb_pack_msgpack_to_json_format(event_chunk->data,
-                                                event_chunk->size,
-                                                FLB_PACK_JSON_FORMAT_LINES,
-                                                ctx->json_date_format,
-                                                ctx->date_key,
-                                                config->json_escape_unicode);
-    }
+    chunk = s3_format_event_chunk(ctx, event_chunk, config);
     if (chunk == NULL) {
         flb_plg_error(ctx->ins, "Could not marshal msgpack to output string");
         FLB_OUTPUT_RETURN(FLB_ERROR);
@@ -3999,6 +4061,11 @@ static int cb_s3_exit(void *data, struct flb_config *config)
 
 /* Configuration properties map */
 static struct flb_config_map config_map[] = {
+    {
+     FLB_CONFIG_MAP_STR, "format", "json_lines",
+     0, FLB_FALSE, 0,
+     "Set record output format. Supported values are json_lines and otlp_json."
+    },
     {
      FLB_CONFIG_MAP_STR, "json_date_format", NULL,
      0, FLB_FALSE, 0,

plugins/out_s3/s3.h

@@ -153,6 +153,7 @@ struct flb_s3 {
     struct flb_tls *client_tls;
 
     struct flb_aws_client *s3_client;
+    int out_format;
     int json_date_format;
     flb_sds_t json_date_key;
     flb_sds_t date_key;