1 /* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
2
3 /* Fluent Bit
4 * ==========
5 * Copyright (C) 2019-2021 The Fluent Bit Authors
6 * Copyright (C) 2015-2018 Treasure Data Inc.
7 *
8 * Licensed under the Apache License, Version 2.0 (the "License");
9 * you may not use this file except in compliance with the License.
10 * You may obtain a copy of the License at
11 *
12 * http://www.apache.org/licenses/LICENSE-2.0
13 *
14 * Unless required by applicable law or agreed to in writing, software
15 * distributed under the License is distributed on an "AS IS" BASIS,
16 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17 * See the License for the specific language governing permissions and
18 * limitations under the License.
19 */
20
21 #include <fluent-bit/flb_output_plugin.h>
22 #include <fluent-bit/flb_network.h>
23 #include <fluent-bit/flb_pack.h>
24 #include <fluent-bit/flb_http_client.h>
25 #include <fluent-bit/flb_time.h>
26 #include <msgpack.h>
27
28 #include "td.h"
29 #include "td_http.h"
30 #include "td_config.h"
31
32 #include <stdio.h>
33 #include <stdlib.h>
34 #include <assert.h>
35 #include <errno.h>
36
37 /*
38 * Convert the internal Fluent Bit data representation to the required
39 * one by Treasure Data cloud service.
40 *
41 * This function returns a new msgpack buffer and store the bytes length
42 * in the out_size variable.
43 */
td_format(const void * data,size_t bytes,int * out_size)44 static char *td_format(const void *data, size_t bytes, int *out_size)
45 {
46 int i;
47 int ret;
48 int n_size;
49 size_t off = 0;
50 time_t atime;
51 char *buf;
52 struct msgpack_sbuffer mp_sbuf;
53 struct msgpack_packer mp_pck;
54 msgpack_unpacked result;
55 msgpack_object root;
56 msgpack_object map;
57 msgpack_sbuffer *sbuf;
58 msgpack_object *obj;
59 struct flb_time tm;
60
61 /* Initialize contexts for new output */
62 msgpack_sbuffer_init(&mp_sbuf);
63 msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
64
65 /* Iterate the original buffer and perform adjustments */
66 msgpack_unpacked_init(&result);
67
68 /* Perform some format validation */
69 ret = msgpack_unpack_next(&result, data, bytes, &off);
70 if (ret == MSGPACK_UNPACK_CONTINUE) {
71 msgpack_unpacked_destroy(&result);
72 return NULL;
73 }
74
75 /* We 'should' get an array */
76 if (result.data.type != MSGPACK_OBJECT_ARRAY) {
77 /*
78 * If we got a different format, we assume the caller knows what he is
79 * doing, we just duplicate the content in a new buffer and cleanup.
80 */
81 buf = flb_malloc(bytes);
82 if (!buf) {
83 flb_errno();
84 msgpack_unpacked_destroy(&result);
85 return NULL;
86 }
87
88 memcpy(buf, data, bytes);
89 *out_size = bytes;
90 msgpack_unpacked_destroy(&result);
91 return buf;
92 }
93
94 root = result.data;
95 if (root.via.array.size == 0) {
96 msgpack_unpacked_destroy(&result);
97 return NULL;
98 }
99
100 off = 0;
101 msgpack_unpacked_destroy(&result);
102 msgpack_unpacked_init(&result);
103 while (msgpack_unpack_next(&result, data, bytes, &off) == MSGPACK_UNPACK_SUCCESS) {
104 if (result.data.type != MSGPACK_OBJECT_ARRAY) {
105 continue;
106 }
107
108 /* Each array must have two entries: time and record */
109 root = result.data;
110 if (root.via.array.size != 2) {
111 continue;
112 }
113
114 flb_time_pop_from_msgpack(&tm, &result, &obj);
115
116 atime = tm.tm.tv_sec;
117 map = root.via.array.ptr[1];
118
119 n_size = map.via.map.size + 1;
120 msgpack_pack_map(&mp_pck, n_size);
121 msgpack_pack_str(&mp_pck, 4);
122 msgpack_pack_str_body(&mp_pck, "time", 4);
123 msgpack_pack_int32(&mp_pck, atime);
124
125 for (i = 0; i < n_size - 1; i++) {
126 msgpack_pack_object(&mp_pck, map.via.map.ptr[i].key);
127 msgpack_pack_object(&mp_pck, map.via.map.ptr[i].val);
128 }
129 }
130 msgpack_unpacked_destroy(&result);
131
132 /* Create new buffer */
133 sbuf = &mp_sbuf;
134 *out_size = sbuf->size;
135 buf = flb_malloc(sbuf->size);
136 if (!buf) {
137 flb_errno();
138 return NULL;
139 }
140
141 /* set a new buffer and re-initialize our MessagePack context */
142 memcpy(buf, sbuf->data, sbuf->size);
143 msgpack_sbuffer_destroy(&mp_sbuf);
144
145 return buf;
146 }
147
cb_td_init(struct flb_output_instance * ins,struct flb_config * config,void * data)148 static int cb_td_init(struct flb_output_instance *ins, struct flb_config *config,
149 void *data)
150 {
151 struct flb_td *ctx;
152 struct flb_upstream *upstream;
153 (void) data;
154
155 ctx = td_config_init(ins);
156 if (!ctx) {
157 flb_plg_warn(ins, "Error reading configuration");
158 return -1;
159 }
160
161 if (ctx->region == FLB_TD_REGION_US) {
162 flb_output_net_default("api.treasuredata.com", 443, ins);
163 }
164 else if (ctx->region == FLB_TD_REGION_JP) {
165 flb_output_net_default("api.treasuredata.co.jp", 443, ins);
166 }
167
168 upstream = flb_upstream_create(config,
169 ins->host.name,
170 ins->host.port,
171 FLB_IO_TLS, ins->tls);
172 if (!upstream) {
173 flb_free(ctx);
174 return -1;
175 }
176 ctx->u = upstream;
177 flb_output_upstream_set(ctx->u, ins);
178
179 flb_output_set_context(ins, ctx);
180 return 0;
181 }
182
cb_td_flush(const void * data,size_t bytes,const char * tag,int tag_len,struct flb_input_instance * i_ins,void * out_context,struct flb_config * config)183 static void cb_td_flush(const void *data, size_t bytes,
184 const char *tag, int tag_len,
185 struct flb_input_instance *i_ins,
186 void *out_context,
187 struct flb_config *config)
188 {
189 int ret;
190 int bytes_out;
191 char *pack;
192 size_t b_sent;
193 char *body = NULL;
194 struct flb_td *ctx = out_context;
195 struct flb_upstream_conn *u_conn;
196 struct flb_http_client *c;
197 (void) i_ins;
198 (void) tag;
199 (void) tag_len;
200
201 /* Convert format */
202 pack = td_format(data, bytes, &bytes_out);
203 if (!pack) {
204 FLB_OUTPUT_RETURN(FLB_ERROR);
205 }
206
207 /* Lookup an available connection context */
208 u_conn = flb_upstream_conn_get(ctx->u);
209 if (!u_conn) {
210 flb_plg_error(ctx->ins, "no upstream connections available");
211 flb_free(pack);
212 FLB_OUTPUT_RETURN(FLB_RETRY);
213 }
214
215 /* Compose request */
216 c = td_http_client(u_conn, pack, bytes_out, &body, ctx, config);
217 if (!c) {
218 flb_free(pack);
219 flb_upstream_conn_release(u_conn);
220 FLB_OUTPUT_RETURN(FLB_RETRY);
221 }
222
223 /* Issue HTTP request */
224 ret = flb_http_do(c, &b_sent);
225
226 /* Release Resources */
227 flb_free(pack);
228 flb_free(body);
229
230 /* Validate HTTP status */
231 if (ret == 0) {
232 /* We expect a HTTP 200 OK */
233 if (c->resp.status != 200) {
234 if (c->resp.payload_size > 0) {
235 flb_plg_warn(ctx->ins, "HTTP status %i\n%s",
236 c->resp.status, c->resp.payload);
237 }
238 else {
239 flb_plg_warn(ctx->ins, "HTTP status %i", c->resp.status);
240 }
241 goto retry;
242 }
243 else {
244 flb_plg_info(ctx->ins, "HTTP status 200 OK");
245 }
246 }
247 else {
248 flb_plg_error(ctx->ins, "http_do=%i", ret);
249 goto retry;
250 }
251
252 /* release */
253 flb_upstream_conn_release(u_conn);
254 flb_http_client_destroy(c);
255
256 FLB_OUTPUT_RETURN(FLB_OK);
257
258 retry:
259 flb_upstream_conn_release(u_conn);
260 flb_http_client_destroy(c);
261
262 FLB_OUTPUT_RETURN(FLB_RETRY);
263 }
264
cb_td_exit(void * data,struct flb_config * config)265 static int cb_td_exit(void *data, struct flb_config *config)
266 {
267 struct flb_td *ctx = data;
268
269 if (!ctx) {
270 return 0;
271 }
272
273 flb_upstream_destroy(ctx->u);
274 flb_free(ctx);
275
276 return 0;
277 }
278
279 /* Plugin reference */
280 struct flb_output_plugin out_td_plugin = {
281 .name = "td",
282 .description = "Treasure Data",
283 .cb_init = cb_td_init,
284 .cb_pre_run = NULL,
285 .cb_flush = cb_td_flush,
286 .cb_exit = cb_td_exit,
287 .flags = FLB_IO_TLS,
288 };
289