blob: 75d41e10718610dfaf8dafa7ac7dd58145733d12 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
|
/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
/* Fluent Bit
* ==========
* Copyright (C) 2015-2022 The Fluent Bit Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#ifndef FLB_OUT_KINESIS_H
#define FLB_OUT_KINESIS_H
#include <fluent-bit/flb_info.h>
#include <fluent-bit/flb_sds.h>
#include <fluent-bit/flb_aws_credentials.h>
#include <fluent-bit/flb_http_client.h>
#include <fluent-bit/flb_aws_util.h>
#include <fluent-bit/flb_signv4.h>
#define DEFAULT_TIME_KEY_FORMAT "%Y-%m-%dT%H:%M:%S"
/* buffers used for each flush */
struct flush {
/* temporary buffer for storing the serialized event messages */
char *tmp_buf;
size_t tmp_buf_size;
/* current index of tmp_buf */
size_t tmp_buf_offset;
/* projected final size of the payload for this flush */
size_t data_size;
/* log records- each of these has a pointer to their message in tmp_buf */
struct kinesis_event *events;
int events_capacity;
/* current event */
int event_index;
/* the payload of the API request */
char *out_buf;
size_t out_buf_size;
/* buffer used to temporarily hold an event during processing */
char *event_buf;
size_t event_buf_size;
int records_sent;
int records_processed;
const char *tag;
int tag_len;
};
struct kinesis_event {
char *json;
size_t len;
struct timespec timestamp;
};
struct flb_kinesis {
/*
* TLS instances can not be re-used. So we have one for:
* - Base cred provider (needed for EKS provider)
* - STS Assume role provider
* - The CloudWatch Logs client for this plugin
*/
struct flb_tls *cred_tls;
struct flb_tls *sts_tls;
struct flb_tls *client_tls;
struct flb_aws_provider *aws_provider;
struct flb_aws_provider *base_aws_provider;
struct flb_aws_client *kinesis_client;
/* configuration options */
const char *stream_name;
const char *time_key;
const char *time_key_format;
const char *region;
const char *role_arn;
const char *log_key;
const char *external_id;
int retry_requests;
char *sts_endpoint;
int custom_endpoint;
char *profile;
/* in this plugin the 'random' partition key is a uuid + fluent tag + timestamp */
char *uuid;
/* must be freed on shutdown if custom_endpoint is not set */
char *endpoint;
/* Plugin output instance reference */
struct flb_output_instance *ins;
};
void flb_kinesis_ctx_destroy(struct flb_kinesis *ctx);
#endif
|
