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 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234
|
/*
* Copyright (c) 2002-2012 Balabit
* Copyright (c) 1998-2012 Balázs Scheidler
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*
* As an additional exemption you are allowed to compile & link against the
* OpenSSL libraries as published by the OpenSSL project. See the file
* COPYING for details.
*
*/
#include "logproto-server.h"
#include "messages.h"
#include "cfg.h"
#include "plugin.h"
#include "plugin-types.h"
#include "ack-tracker/ack_tracker_factory.h"
/**
* Find the character terminating the buffer.
*
* NOTE: when looking for the end-of-message here, it either needs to be
* terminated via NUL or via NL, when terminating via NL we have to make
* sure that there's no NUL left in the message. This function iterates over
* the input data and returns a pointer to the first occurrence of NL or NUL.
*
* It uses an algorithm similar to what there's in libc memchr/strchr.
*
* NOTE: find_eom is not static as it is used by a unit test program.
**/
const guchar *
find_eom(const guchar *s, gsize n)
{
const guchar *char_ptr;
const gulong *longword_ptr;
gulong longword, magic_bits, charmask;
gchar c;
c = '\n';
/* align input to long boundary */
for (char_ptr = s; n > 0 && ((gulong) char_ptr & (sizeof(longword) - 1)) != 0; ++char_ptr, n--)
{
if (*char_ptr == c || *char_ptr == '\0')
return char_ptr;
}
longword_ptr = (gulong *) char_ptr;
#if GLIB_SIZEOF_LONG == 8
magic_bits = 0x7efefefefefefeffL;
#elif GLIB_SIZEOF_LONG == 4
magic_bits = 0x7efefeffL;
#else
#error "unknown architecture"
#endif
memset(&charmask, c, sizeof(charmask));
while (n > sizeof(longword))
{
longword = *longword_ptr++;
if ((((longword + magic_bits) ^ ~longword) & ~magic_bits) != 0 ||
((((longword ^ charmask) + magic_bits) ^ ~(longword ^ charmask)) & ~magic_bits) != 0)
{
gint i;
char_ptr = (const guchar *) (longword_ptr - 1);
for (i = 0; i < sizeof(longword); i++)
{
if (*char_ptr == c || *char_ptr == '\0')
return char_ptr;
char_ptr++;
}
}
n -= sizeof(longword);
}
char_ptr = (const guchar *) longword_ptr;
while (n-- > 0)
{
if (*char_ptr == c || *char_ptr == '\0')
return char_ptr;
++char_ptr;
}
return NULL;
}
AckTrackerFactory *
log_proto_server_get_ack_tracker_factory(LogProtoServer *s)
{
return s->options->ack_tracker_factory;
}
gboolean
log_proto_server_is_position_tracked(LogProtoServer *s)
{
AckTrackerType type = ack_tracker_factory_get_type(log_proto_server_get_ack_tracker_factory(s));
return ack_tracker_type_is_position_tracked(type);
}
gboolean
log_proto_server_validate_options_method(LogProtoServer *s)
{
return TRUE;
}
void
log_proto_server_free_method(LogProtoServer *s)
{
log_transport_free(s->transport);
}
void
log_proto_server_free(LogProtoServer *s)
{
if (s->free_fn)
s->free_fn(s);
g_free(s);
}
void
log_proto_server_init(LogProtoServer *self, LogTransport *transport, const LogProtoServerOptions *options)
{
self->validate_options = log_proto_server_validate_options_method;
self->free_fn = log_proto_server_free_method;
self->options = options;
self->transport = transport;
}
gboolean
log_proto_server_options_set_encoding(LogProtoServerOptions *self, const gchar *encoding)
{
GIConv convert;
g_free(self->encoding);
self->encoding = g_strdup(encoding);
/* validate encoding */
convert = g_iconv_open("utf-8", encoding);
if (convert == (GIConv) -1)
return FALSE;
g_iconv_close(convert);
return TRUE;
}
void
log_proto_server_options_set_ack_tracker_factory(LogProtoServerOptions *self, AckTrackerFactory *factory)
{
ack_tracker_factory_unref(self->ack_tracker_factory);
self->ack_tracker_factory = factory;
}
void
log_proto_server_options_defaults(LogProtoServerOptions *options)
{
memset(options, 0, sizeof(*options));
options->max_msg_size = -1;
options->trim_large_messages = -1;
options->init_buffer_size = -1;
options->max_buffer_size = -1;
options->ack_tracker_factory = instant_ack_tracker_bookmarkless_factory_new();
}
void
log_proto_server_options_init(LogProtoServerOptions *options, GlobalConfig *cfg)
{
if (options->initialized)
return;
if (options->max_msg_size == -1)
{
options->max_msg_size = cfg->log_msg_size;
}
if (options->trim_large_messages == -1)
{
options->trim_large_messages = cfg->trim_large_messages;
}
if (options->max_buffer_size == -1)
{
if (options->encoding)
{
/* Based on the implementation of LogProtoTextServer, the buffer is yielded as
a complete message when max_msg_size is reached and there is no EOM in the buffer.
In worst case, the buffer contains max_msg_size - 1 bytes before the next fetch,
which can be 6 times max_msg_size due to the utf8 conversion.
And additional space is required because of the possible leftover bytes.
*/
options->max_buffer_size = 8 * options->max_msg_size;
}
else
options->max_buffer_size = options->max_msg_size;
}
if (options->init_buffer_size == -1)
options->init_buffer_size = MIN(options->max_msg_size, options->max_buffer_size);
options->initialized = TRUE;
}
void
log_proto_server_options_destroy(LogProtoServerOptions *options)
{
g_free(options->encoding);
ack_tracker_factory_unref(options->ack_tracker_factory);
if (options->destroy)
options->destroy(options);
options->initialized = FALSE;
}
LogProtoServerFactory *
log_proto_server_get_factory(PluginContext *context, const gchar *name)
{
Plugin *plugin;
plugin = plugin_find(context, LL_CONTEXT_SERVER_PROTO, name);
if (plugin && plugin->construct)
return plugin->construct(plugin);
return NULL;
}
|