db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync#include "journal-remote-parse.h"

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync#include "journald-native.h"

10cdf5733351fdcd857d439ca32189e812f18682vboxsync

10cdf5733351fdcd857d439ca32189e812f18682vboxsync#define LINE_CHUNK 8*1024u

10cdf5733351fdcd857d439ca32189e812f18682vboxsync

10cdf5733351fdcd857d439ca32189e812f18682vboxsyncvoid source_free(RemoteSource *source) {

10cdf5733351fdcd857d439ca32189e812f18682vboxsync if (!source)

10cdf5733351fdcd857d439ca32189e812f18682vboxsync return;

10cdf5733351fdcd857d439ca32189e812f18682vboxsync

10cdf5733351fdcd857d439ca32189e812f18682vboxsync if (source->fd >= 0 && !source->passive_fd) {

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync log_debug("Closing fd:%d (%s)", source->fd, source->name);

10cdf5733351fdcd857d439ca32189e812f18682vboxsync safe_close(source->fd);

10cdf5733351fdcd857d439ca32189e812f18682vboxsync }

10cdf5733351fdcd857d439ca32189e812f18682vboxsync

10cdf5733351fdcd857d439ca32189e812f18682vboxsync free(source->name);

10cdf5733351fdcd857d439ca32189e812f18682vboxsync free(source->buf);

10cdf5733351fdcd857d439ca32189e812f18682vboxsync iovw_free_contents(&source->iovw);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync log_debug("Writer ref count %u", source->writer->n_ref);

cd9e4940318086a06a68bf301960563dcb72b939vboxsync writer_unref(source->writer);

cd9e4940318086a06a68bf301960563dcb72b939vboxsync

cd9e4940318086a06a68bf301960563dcb72b939vboxsync sd_event_source_unref(source->event);

10cdf5733351fdcd857d439ca32189e812f18682vboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync free(source);

10cdf5733351fdcd857d439ca32189e812f18682vboxsync}

10cdf5733351fdcd857d439ca32189e812f18682vboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsyncRemoteSource* source_new(int fd, bool passive_fd, char *name, Writer *writer) {

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

10cdf5733351fdcd857d439ca32189e812f18682vboxsync RemoteSource *source;

10cdf5733351fdcd857d439ca32189e812f18682vboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync log_debug("Creating source for %sfd:%d (%s)",

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync passive_fd ? "passive " : "", fd, name);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync assert(fd >= 0);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync source = new0(RemoteSource, 1);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync if (!source)

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync return NULL;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

cd9e4940318086a06a68bf301960563dcb72b939vboxsync source->fd = fd;

10cdf5733351fdcd857d439ca32189e812f18682vboxsync source->passive_fd = passive_fd;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync source->name = name;

10cdf5733351fdcd857d439ca32189e812f18682vboxsync source->writer = writer;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync return source;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync}

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsyncstatic char* realloc_buffer(RemoteSource *source, size_t size) {

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync char *b, *old = source->buf;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync b = GREEDY_REALLOC(source->buf, source->size, size);

86abc60770f825f8c2ed4257675b50a08743b687vboxsync if (!b)

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync return NULL;

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync iovw_rebase(&source->iovw, old, source->buf);

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync return b;

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync}

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync

4a429a59b1a82ce092626ea5f7512466c18f2015vboxsyncstatic int get_line(RemoteSource *source, char **line, size_t *size) {

4a429a59b1a82ce092626ea5f7512466c18f2015vboxsync ssize_t n;

4a429a59b1a82ce092626ea5f7512466c18f2015vboxsync char *c = NULL;

4a429a59b1a82ce092626ea5f7512466c18f2015vboxsync

4a429a59b1a82ce092626ea5f7512466c18f2015vboxsync assert(source);

4a429a59b1a82ce092626ea5f7512466c18f2015vboxsync assert(source->state == STATE_LINE);

4a429a59b1a82ce092626ea5f7512466c18f2015vboxsync assert(source->offset <= source->filled);

4a429a59b1a82ce092626ea5f7512466c18f2015vboxsync assert(source->filled <= source->size);

4a429a59b1a82ce092626ea5f7512466c18f2015vboxsync assert(source->buf == NULL || source->size > 0);

4a429a59b1a82ce092626ea5f7512466c18f2015vboxsync assert(source->fd >= 0);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync while (true) {

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync if (source->buf) {

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync size_t start = MAX(source->scanned, source->offset);

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync c = memchr(source->buf + start, '\n',

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync source->filled - start);

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync if (c != NULL)

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync break;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync }

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

86abc60770f825f8c2ed4257675b50a08743b687vboxsync source->scanned = source->filled;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync if (source->scanned >= DATA_SIZE_MAX) {

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync log_error("Entry is bigger than %u bytes.", DATA_SIZE_MAX);

86abc60770f825f8c2ed4257675b50a08743b687vboxsync return -E2BIG;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync }

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync if (source->passive_fd)

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync /* we have to wait for some data to come to us */

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync return -EWOULDBLOCK;

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync if (source->size - source->filled < LINE_CHUNK &&

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync !realloc_buffer(source,

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync MIN(source->filled + LINE_CHUNK, ENTRY_SIZE_MAX)))

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync return log_oom();

360c39d88dfa26b17181b57ebafdb24c2a113c63vboxsync

360c39d88dfa26b17181b57ebafdb24c2a113c63vboxsync assert(source->size - source->filled >= LINE_CHUNK ||

360c39d88dfa26b17181b57ebafdb24c2a113c63vboxsync source->size == ENTRY_SIZE_MAX);

360c39d88dfa26b17181b57ebafdb24c2a113c63vboxsync

360c39d88dfa26b17181b57ebafdb24c2a113c63vboxsync n = read(source->fd, source->buf + source->filled,

360c39d88dfa26b17181b57ebafdb24c2a113c63vboxsync source->size - source->filled);

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync if (n < 0) {

02fb80eb0db13569db21d1ce5e0f3e0d5a6122aavboxsync if (errno != EAGAIN && errno != EWOULDBLOCK)

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync log_error("read(%d, ..., %zd): %m", source->fd,

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync source->size - source->filled);

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync return -errno;

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync } else if (n == 0)

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync return 0;

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync source->filled += n;

6967517de4be849f55b0141d6089add0eff2aa7bvboxsync }

6967517de4be849f55b0141d6089add0eff2aa7bvboxsync

23ee8310386e73ba6760fa30831a7964713d34b6vboxsync *line = source->buf + source->offset;

86abc60770f825f8c2ed4257675b50a08743b687vboxsync *size = c + 1 - source->buf - source->offset;

7b213bb002950f9fcf809f605cc584fa543481advboxsync source->offset += *size;

7b213bb002950f9fcf809f605cc584fa543481advboxsync

7b213bb002950f9fcf809f605cc584fa543481advboxsync return 1;

7b213bb002950f9fcf809f605cc584fa543481advboxsync}

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsyncint push_data(RemoteSource *source, const char *data, size_t size) {

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync assert(source);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync assert(source->state != STATE_EOF);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync if (!realloc_buffer(source, source->filled + size)) {

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync log_error("Failed to store received data of size %zu "

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync "(in addition to existing %zu bytes with %zu filled): %s",

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync size, source->size, source->filled, strerror(ENOMEM));

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync return -ENOMEM;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync }

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

cd9e4940318086a06a68bf301960563dcb72b939vboxsync memcpy(source->buf + source->filled, data, size);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync source->filled += size;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync return 0;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync}

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsyncstatic int fill_fixed_size(RemoteSource *source, void **data, size_t size) {

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync assert(source);

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync assert(source->state == STATE_DATA_START ||

b49952fd6cae765238f4760c700c5ef797a56092vboxsync source->state == STATE_DATA ||

b49952fd6cae765238f4760c700c5ef797a56092vboxsync source->state == STATE_DATA_FINISH);

b49952fd6cae765238f4760c700c5ef797a56092vboxsync assert(size <= DATA_SIZE_MAX);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync assert(source->offset <= source->filled);

cd9e4940318086a06a68bf301960563dcb72b939vboxsync assert(source->filled <= source->size);

8364ffb7e421fa1ec2bd282ab4c52ac718873d46vboxsync assert(source->buf != NULL || source->size == 0);

8364ffb7e421fa1ec2bd282ab4c52ac718873d46vboxsync assert(source->buf == NULL || source->size > 0);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync assert(source->fd >= 0);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync assert(data);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync while (source->filled - source->offset < size) {

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync int n;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync if (source->passive_fd)

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync /* we have to wait for some data to come to us */

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync return -EWOULDBLOCK;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync if (!realloc_buffer(source, source->offset + size))

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync return log_oom();

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync n = read(source->fd, source->buf + source->filled,

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync source->size - source->filled);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync if (n < 0) {

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync if (errno != EAGAIN && errno != EWOULDBLOCK)

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync log_error("read(%d, ..., %zd): %m", source->fd,

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync source->size - source->filled);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync return -errno;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync } else if (n == 0)

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync return 0;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync source->filled += n;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync }

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

c7a92d481aced08517570d43bf75e2a9f3a8aaeavboxsync *data = source->buf + source->offset;

c7a92d481aced08517570d43bf75e2a9f3a8aaeavboxsync source->offset += size;

c7a92d481aced08517570d43bf75e2a9f3a8aaeavboxsync

c7a92d481aced08517570d43bf75e2a9f3a8aaeavboxsync return 1;

c7a92d481aced08517570d43bf75e2a9f3a8aaeavboxsync}

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsyncstatic int get_data_size(RemoteSource *source) {

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync int r;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync void *data;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync assert(source);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync assert(source->state == STATE_DATA_START);

c7a92d481aced08517570d43bf75e2a9f3a8aaeavboxsync assert(source->data_size == 0);

541ba632c438350cc8044d7ce2c8623dca446546vboxsync

08a80484275b5172ce23729ecccc934c6a92d201vboxsync r = fill_fixed_size(source, &data, sizeof(uint64_t));

08a80484275b5172ce23729ecccc934c6a92d201vboxsync if (r <= 0)

08a80484275b5172ce23729ecccc934c6a92d201vboxsync return r;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

86abc60770f825f8c2ed4257675b50a08743b687vboxsync source->data_size = le64toh( *(uint64_t *) data );

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync if (source->data_size > DATA_SIZE_MAX) {

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync log_error("Stream declares field with size %zu > DATA_SIZE_MAX = %u",

48f33dfd8f615d457106bf76ae2d09b8b9167c1avboxsync source->data_size, DATA_SIZE_MAX);

48f33dfd8f615d457106bf76ae2d09b8b9167c1avboxsync return -EINVAL;

48f33dfd8f615d457106bf76ae2d09b8b9167c1avboxsync }

48f33dfd8f615d457106bf76ae2d09b8b9167c1avboxsync if (source->data_size == 0)

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync log_warning("Binary field with zero length");

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync return 1;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync}

fc148a6b23d25a87561beaffe0ba06c3ba93bf5avboxsync

fc148a6b23d25a87561beaffe0ba06c3ba93bf5avboxsyncstatic int get_data_data(RemoteSource *source, void **data) {

fc148a6b23d25a87561beaffe0ba06c3ba93bf5avboxsync int r;

fc148a6b23d25a87561beaffe0ba06c3ba93bf5avboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync assert(source);

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync assert(data);

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync assert(source->state == STATE_DATA);

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync r = fill_fixed_size(source, data, source->data_size);

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync if (r <= 0)

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync return r;

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync return 1;

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync}

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsyncstatic int get_data_newline(RemoteSource *source) {

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync int r;

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync char *data;

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync assert(source);

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync assert(source->state == STATE_DATA_FINISH);

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync r = fill_fixed_size(source, (void**) &data, 1);

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync if (r <= 0)

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync return r;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync assert(data);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync if (*data != '\n') {

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync log_error("expected newline, got '%c'", *data);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync return -EINVAL;

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync }

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync

bc8beda7587b16cafe7de7bea19149de1bd4b498vboxsync return 1;

bc8beda7587b16cafe7de7bea19149de1bd4b498vboxsync}

59e30364f54880dd846c263711a2506d1182b1b5vboxsync

59e30364f54880dd846c263711a2506d1182b1b5vboxsyncstatic int process_dunder(RemoteSource *source, char *line, size_t n) {

59e30364f54880dd846c263711a2506d1182b1b5vboxsync const char *timestamp;

59e30364f54880dd846c263711a2506d1182b1b5vboxsync int r;

3acaac88e9a4c925226db26f0d374f9f4876b74fvboxsync

cc8517c11be66037a9873fa03f280e7742efed6dvboxsync assert(line);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync assert(n > 0);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync assert(line[n-1] == '\n');

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync /* XXX: is it worth to support timestamps in extended format?

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync timestamp = startswith(line, "__CURSOR=");

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync if (timestamp)

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync /* ignore __CURSOR */

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync return 1;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync timestamp = startswith(line, "__REALTIME_TIMESTAMP=");

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync if (timestamp) {

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync long long unsigned x;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync line[n-1] = '\0';

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync r = safe_atollu(timestamp, &x);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync if (r < 0)

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync log_warning("Failed to parse __REALTIME_TIMESTAMP: '%s'", timestamp);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync else

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync source->ts.realtime = x;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync return r < 0 ? r : 1;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync }

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync timestamp = startswith(line, "__MONOTONIC_TIMESTAMP=");

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync if (timestamp) {

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync long long unsigned x;

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync line[n-1] = '\0';

541ba632c438350cc8044d7ce2c8623dca446546vboxsync r = safe_atollu(timestamp, &x);

08a56d5836eceeb24642b61eaa52a4edb0a7b482vboxsync if (r < 0)

08a56d5836eceeb24642b61eaa52a4edb0a7b482vboxsync log_warning("Failed to parse __MONOTONIC_TIMESTAMP: '%s'", timestamp);

08a56d5836eceeb24642b61eaa52a4edb0a7b482vboxsync else

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync source->ts.monotonic = x;

3b3bc8a9383a065307e540b83fc3a3d6c548a082vboxsync return r < 0 ? r : 1;

48f33dfd8f615d457106bf76ae2d09b8b9167c1avboxsync }

48f33dfd8f615d457106bf76ae2d09b8b9167c1avboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync timestamp = startswith(line, "__");

48f33dfd8f615d457106bf76ae2d09b8b9167c1avboxsync if (timestamp) {

48f33dfd8f615d457106bf76ae2d09b8b9167c1avboxsync log_notice("Unknown dunder line %s", line);

48f33dfd8f615d457106bf76ae2d09b8b9167c1avboxsync return 1;

48f33dfd8f615d457106bf76ae2d09b8b9167c1avboxsync }

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync /* no dunder */

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync return 0;

3196a540aa11c41264b0e79857b879e4ab566f4bvboxsync}

3196a540aa11c41264b0e79857b879e4ab566f4bvboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsyncint process_data(RemoteSource *source) {

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync int r;

3196a540aa11c41264b0e79857b879e4ab566f4bvboxsync

3196a540aa11c41264b0e79857b879e4ab566f4bvboxsync switch(source->state) {

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync case STATE_LINE: {

16aeb8f83239d4a4a10ecac1fb46fe24a8bdbb66vboxsync char *line, *sep;

16aeb8f83239d4a4a10ecac1fb46fe24a8bdbb66vboxsync size_t n;

16aeb8f83239d4a4a10ecac1fb46fe24a8bdbb66vboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync assert(source->data_size == 0);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync r = get_line(source, &line, &n);

2705a216ac12110a0813d671e230797c886b4788vboxsync if (r < 0)

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync return r;

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync if (r == 0) {

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync source->state = STATE_EOF;

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync return r;

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync }

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync assert(n > 0);

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync assert(line[n-1] == '\n');

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync if (n == 1) {

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync log_debug("Received empty line, event is ready");

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync return 1;

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync }

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync r = process_dunder(source, line, n);

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync if (r != 0)

24b88f881835a6c392e72177f74f1d5e4544ba1evboxsync return r < 0 ? r : 0;

f8244da4b4e02d8d4ce0669eeb4093e31c301888vboxsync

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync /* MESSAGE=xxx\n

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync sep = memchr(line, '=', n);

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync if (sep)

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync /* chomp newline */

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync n--;

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync else

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync /* replace \n with = */

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync line[n-1] = '=';

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync log_debug("Received: %.*s", (int) n, line);

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync r = iovw_put(&source->iovw, line, n);

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync if (r < 0) {

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync log_error("Failed to put line in iovect");

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync return r;

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync }

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync if (!sep)

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync source->state = STATE_DATA_START;

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync return 0; /* continue */

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync }

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync case STATE_DATA_START:

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync assert(source->data_size == 0);

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync r = get_data_size(source);

bee07f28c146fb5313ea13d84dcd508d8f604a10vboxsync log_debug("get_data_size() -> %d", r);

441e51789084e8795573747fec9f5108e8b54e98vboxsync if (r < 0)

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync return r;

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync if (r == 0) {

15ca5f803f0c1bfc86a224523ef482f8aa960388vboxsync source->state = STATE_EOF;

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync return 0;

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync }

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync source->state = source->data_size > 0 ?

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync STATE_DATA : STATE_DATA_FINISH;

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync return 0; /* continue */

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync case STATE_DATA: {

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync void *data;

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync assert(source->data_size > 0);

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync r = get_data_data(source, &data);

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync log_debug("get_data_data() -> %d", r);

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync if (r < 0)

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync return r;

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync if (r == 0) {

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync source->state = STATE_EOF;

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync return 0;

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync }

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync assert(data);

801b06499d776aed19ddc7755183ea2d39de20eavboxsync

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync r = iovw_put(&source->iovw, data, source->data_size);

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync if (r < 0) {

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync log_error("failed to put binary buffer in iovect");

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync return r;

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync }

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync source->state = STATE_DATA_FINISH;

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync return 0; /* continue */

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync }

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync case STATE_DATA_FINISH:

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync r = get_data_newline(source);

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync log_debug("get_data_newline() -> %d", r);

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync if (r < 0)

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync return r;

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync if (r == 0) {

8d466f9285d86e81f927c2bf053a2eb7ec325746vboxsync source->state = STATE_EOF;

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync return 0;

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync }

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync source->data_size = 0;

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync source->state = STATE_LINE;

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync

1cc3bd5463294790ba54c78fde5313264185e50cvboxsync return 0; /* continue */

efda5c4c4db213abd0692df0ea34a26f6230d59avboxsync default:

2c76fd8cdec6f32663a03b16fb5871f7c54330a6vboxsync assert_not_reached("wtf?");

2c76fd8cdec6f32663a03b16fb5871f7c54330a6vboxsync }

2705a216ac12110a0813d671e230797c886b4788vboxsync}

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsyncint process_source(RemoteSource *source, bool compress, bool seal) {

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync size_t remain, target;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync int r;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync assert(source);

c7a92d481aced08517570d43bf75e2a9f3a8aaeavboxsync assert(source->writer);

541ba632c438350cc8044d7ce2c8623dca446546vboxsync

42d36464aee6c81b4f206b6c02035587501ff67cvboxsync r = process_data(source);

c3e38cccf650831700227918a021e6c4097ace82vboxsync if (r <= 0)

42d36464aee6c81b4f206b6c02035587501ff67cvboxsync return r;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync /* We have a full event */

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync log_debug("Received a full event from source@%p fd:%d (%s)",

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync source, source->fd, source->name);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

3b3bc8a9383a065307e540b83fc3a3d6c548a082vboxsync if (!source->iovw.count) {

48f33dfd8f615d457106bf76ae2d09b8b9167c1avboxsync log_warning("Entry with no payload, skipping");

48f33dfd8f615d457106bf76ae2d09b8b9167c1avboxsync goto freeing;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync }

48f33dfd8f615d457106bf76ae2d09b8b9167c1avboxsync

48f33dfd8f615d457106bf76ae2d09b8b9167c1avboxsync assert(source->iovw.iovec);

48f33dfd8f615d457106bf76ae2d09b8b9167c1avboxsync assert(source->iovw.count);

48f33dfd8f615d457106bf76ae2d09b8b9167c1avboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync r = writer_write(source->writer, &source->iovw, &source->ts, compress, seal);

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync if (r < 0)

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync log_error("Failed to write entry of %zu bytes: %s",

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync iovw_size(&source->iovw), strerror(-r));

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync else

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync r = 1;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync freeing:

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync iovw_free_contents(&source->iovw);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync /* possibly reset buffer position */

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync remain = source->filled - source->offset;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync if (remain == 0) /* no brainer */

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync source->offset = source->scanned = source->filled = 0;

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync else if (source->offset > source->size - source->filled &&

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync source->offset > remain) {

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync memcpy(source->buf, source->buf + source->offset, remain);

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync source->offset = source->scanned = 0;

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync source->filled = remain;

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync }

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync target = source->size;

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync while (target > 16 * LINE_CHUNK && remain < target / 2)

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync target /= 2;

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync if (target < source->size) {

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync char *tmp;

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync tmp = realloc(source->buf, target);

95d42763b8808d795c23148d7dbc00a3b7b40d6fvboxsync if (tmp)

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync log_warning("Failed to reallocate buffer to (smaller) size %zu",

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync target);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync else {

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync log_debug("Reallocated buffer from %zu to %zu bytes",

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync source->size, target);

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync source->buf = tmp;

cd9e4940318086a06a68bf301960563dcb72b939vboxsync source->size = target;

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync }

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync }

5857f4e58ce2ef50d7f0c450fe4897026f9a9c3dvboxsync

42dc09ee69e746b8641cfa190931a15ecfd7295cvboxsync return r;

e0b91e4f93fb43371374df4aeb636dffea336056vboxsync}

db3dbd0ed7eb69f804a8921fa23a1267ea01f46evboxsync