<commit_before>// +build linux,cgo,!static_build,journald

package journald

// #include <sys/types.h>
// #include <sys/poll.h>
// #include <systemd/sd-journal.h>
// #include <errno.h>
// #include <stdio.h>
// #include <stdlib.h>
// #include <string.h>
// #include <time.h>
// #include <unistd.h>
//
//static int get_message(sd_journal *j, const char **msg, size_t *length)
//{
//	int rc;
//	*msg = NULL;
//	*length = 0;
//	rc = sd_journal_get_data(j, "MESSAGE", (const void **) msg, length);
//	if (rc == 0) {
//		if (*length > 8) {
//			(*msg) += 8;
//			*length -= 8;
//		} else {
//			*msg = NULL;
//			*length = 0;
//			rc = -ENOENT;
//		}
//	}
//	return rc;
//}
//static int get_priority(sd_journal *j, int *priority)
//{
//	const void *data;
//	size_t i, length;
//	int rc;
//	*priority = -1;
//	rc = sd_journal_get_data(j, "PRIORITY", &data, &length);
//	if (rc == 0) {
//		if ((length > 9) && (strncmp(data, "PRIORITY=", 9) == 0)) {
//			*priority = 0;
//			for (i = 9; i < length; i++) {
//				*priority = *priority * 10 + ((const char *)data)[i] - '0';
//			}
//			if (length > 9) {
//				rc = 0;
//			}
//		}
//	}
//	return rc;
//}
//static int wait_for_data_or_close(sd_journal *j, int pipefd)
//{
//	struct pollfd fds[2];
//	uint64_t when = 0;
//	int timeout, jevents, i;
//	struct timespec ts;
//	uint64_t now;
//	do {
//		memset(&fds, 0, sizeof(fds));
//		fds[0].fd = pipefd;
//		fds[0].events = POLLHUP;
//		fds[1].fd = sd_journal_get_fd(j);
//		if (fds[1].fd < 0) {
//			return -1;
//		}
//		jevents = sd_journal_get_events(j);
//		if (jevents < 0) {
//			return -1;
//		}
//		fds[1].events = jevents;
//		sd_journal_get_timeout(j, &when);
//		if (when == -1) {
//			timeout = -1;
//		} else {
//			clock_gettime(CLOCK_MONOTONIC, &ts);
//			now = (uint64_t) ts.tv_sec * 1000000 + ts.tv_nsec / 1000;
//			timeout = when > now ? (int) ((when - now + 999) / 1000) : 0;
//		}
//		i = poll(fds, 2, timeout);
//		if ((i == -1) && (errno != EINTR)) {
//			/* An unexpected error. */
//			return -1;
//		}
//		if (fds[0].revents & POLLHUP) {
//			/* The close notification pipe was closed. */
//			return 0;
//		}
//		if (sd_journal_process(j) == SD_JOURNAL_APPEND) {
//			/* Data, which we might care about, was appended. */
//			return 1;
//		}
//	} while ((fds[0].revents & POLLHUP) == 0);
//	return 0;
//}
import "C"

import (
	"fmt"
	"time"
	"unsafe"

	"github.com/coreos/go-systemd/journal"
	"github.com/docker/docker/daemon/logger"
)

func (s *journald) Close() error {
	s.readers.mu.Lock()
	for reader := range s.readers.readers {
		reader.Close()
	}
	s.readers.mu.Unlock()
	return nil
}

func (s *journald) drainJournal(logWatcher *logger.LogWatcher, config logger.ReadConfig, j *C.sd_journal, oldCursor string) string {
	var msg, cursor *C.char
	var length C.size_t
	var stamp C.uint64_t
	var priority C.int

	// Walk the journal from here forward until we run out of new entries.
drain:
	for {
		// Try not to send a given entry twice.
		if oldCursor != "" {
			ccursor := C.CString(oldCursor)
			defer C.free(unsafe.Pointer(ccursor))
			for C.sd_journal_test_cursor(j, ccursor) > 0 {
				if C.sd_journal_next(j) <= 0 {
					break drain
				}
			}
		}
		// Read and send the logged message, if there is one to read.
		i := C.get_message(j, &msg, &length)
		if i != -C.ENOENT && i != -C.EADDRNOTAVAIL {
			// Read the entry's timestamp.
			if C.sd_journal_get_realtime_usec(j, &stamp) != 0 {
				break
			}
			// Set up the time and text of the entry.
			timestamp := time.Unix(int64(stamp)/1000000, (int64(stamp)%1000000)*1000)
			line := append(C.GoBytes(unsafe.Pointer(msg), C.int(length)), "\n"...)
			// Recover the stream name by mapping
			// from the journal priority back to
			// the stream that we would have
			// assigned that value.
			source := ""
			if C.get_priority(j, &priority) != 0 {
				source = ""
			} else if priority == C.int(journal.PriErr) {
				source = "stderr"
			} else if priority == C.int(journal.PriInfo) {
				source = "stdout"
			}
			// Send the log message.
			cid := s.vars["CONTAINER_ID_FULL"]
			logWatcher.Msg <- &logger.Message{ContainerID: cid, Line: line, Source: source, Timestamp: timestamp}
		}
		// If we're at the end of the journal, we're done (for now).
		if C.sd_journal_next(j) <= 0 {
			break
		}
	}
	retCursor := ""
	if C.sd_journal_get_cursor(j, &cursor) == 0 {
		retCursor = C.GoString(cursor)
		C.free(unsafe.Pointer(cursor))
	}
	return retCursor
}

func (s *journald) followJournal(logWatcher *logger.LogWatcher, config logger.ReadConfig, j *C.sd_journal, pfd [2]C.int, cursor string) {
	s.readers.mu.Lock()
	s.readers.readers[logWatcher] = logWatcher
	s.readers.mu.Unlock()
	go func() {
		// Keep copying journal data out until we're notified to stop.
		for C.wait_for_data_or_close(j, pfd[0]) == 1 {
			cursor = s.drainJournal(logWatcher, config, j, cursor)
		}
		// Clean up.
		C.close(pfd[0])
		s.readers.mu.Lock()
		delete(s.readers.readers, logWatcher)
		s.readers.mu.Unlock()
		C.sd_journal_close(j)
		close(logWatcher.Msg)
	}()
	// Wait until we're told to stop.
	select {
	case <-logWatcher.WatchClose():
		// Notify the other goroutine that its work is done.
		C.close(pfd[1])
	}
}

func (s *journald) readLogs(logWatcher *logger.LogWatcher, config logger.ReadConfig) {
	var j *C.sd_journal
	var cmatch *C.char
	var stamp C.uint64_t
	var sinceUnixMicro uint64
	var pipes [2]C.int
	cursor := ""
<commit_msg>Improve error reporting when following journals<commit_after>// +build linux,cgo,!static_build,journald

package journald

// #include <sys/types.h>
// #include <sys/poll.h>
// #include <systemd/sd-journal.h>
// #include <errno.h>
// #include <stdio.h>
// #include <stdlib.h>
// #include <string.h>
// #include <time.h>
// #include <unistd.h>
//
//static int get_message(sd_journal *j, const char **msg, size_t *length)
//{
//	int rc;
//	*msg = NULL;
//	*length = 0;
//	rc = sd_journal_get_data(j, "MESSAGE", (const void **) msg, length);
//	if (rc == 0) {
//		if (*length > 8) {
//			(*msg) += 8;
//			*length -= 8;
//		} else {
//			*msg = NULL;
//			*length = 0;
//			rc = -ENOENT;
//		}
//	}
//	return rc;
//}
//static int get_priority(sd_journal *j, int *priority)
//{
//	const void *data;
//	size_t i, length;
//	int rc;
//	*priority = -1;
//	rc = sd_journal_get_data(j, "PRIORITY", &data, &length);
//	if (rc == 0) {
//		if ((length > 9) && (strncmp(data, "PRIORITY=", 9) == 0)) {
//			*priority = 0;
//			for (i = 9; i < length; i++) {
//				*priority = *priority * 10 + ((const char *)data)[i] - '0';
//			}
//			if (length > 9) {
//				rc = 0;
//			}
//		}
//	}
//	return rc;
//}
//static int wait_for_data_or_close(sd_journal *j, int pipefd)
//{
//	struct pollfd fds[2];
//	uint64_t when = 0;
//	int timeout, jevents, i;
//	struct timespec ts;
//	uint64_t now;
//	do {
//		memset(&fds, 0, sizeof(fds));
//		fds[0].fd = pipefd;
//		fds[0].events = POLLHUP;
//		fds[1].fd = sd_journal_get_fd(j);
//		if (fds[1].fd < 0) {
//			return fds[1].fd;
//		}
//		jevents = sd_journal_get_events(j);
//		if (jevents < 0) {
//			return jevents;
//		}
//		fds[1].events = jevents;
//		sd_journal_get_timeout(j, &when);
//		if (when == -1) {
//			timeout = -1;
//		} else {
//			clock_gettime(CLOCK_MONOTONIC, &ts);
//			now = (uint64_t) ts.tv_sec * 1000000 + ts.tv_nsec / 1000;
//			timeout = when > now ? (int) ((when - now + 999) / 1000) : 0;
//		}
//		i = poll(fds, 2, timeout);
//		if ((i == -1) && (errno != EINTR)) {
//			/* An unexpected error. */
//			return (errno != 0) ? -errno : -EINTR;
//		}
//		if (fds[0].revents & POLLHUP) {
//			/* The close notification pipe was closed. */
//			return 0;
//		}
//		if (sd_journal_process(j) == SD_JOURNAL_APPEND) {
//			/* Data, which we might care about, was appended. */
//			return 1;
//		}
//	} while ((fds[0].revents & POLLHUP) == 0);
//	return 0;
//}
import "C"

import (
	"fmt"
	"time"
	"unsafe"

	"github.com/Sirupsen/logrus"
	"github.com/coreos/go-systemd/journal"
	"github.com/docker/docker/daemon/logger"
)

func (s *journald) Close() error {
	s.readers.mu.Lock()
	for reader := range s.readers.readers {
		reader.Close()
	}
	s.readers.mu.Unlock()
	return nil
}

func (s *journald) drainJournal(logWatcher *logger.LogWatcher, config logger.ReadConfig, j *C.sd_journal, oldCursor string) string {
	var msg, cursor *C.char
	var length C.size_t
	var stamp C.uint64_t
	var priority C.int

	// Walk the journal from here forward until we run out of new entries.
drain:
	for {
		// Try not to send a given entry twice.
		if oldCursor != "" {
			ccursor := C.CString(oldCursor)
			defer C.free(unsafe.Pointer(ccursor))
			for C.sd_journal_test_cursor(j, ccursor) > 0 {
				if C.sd_journal_next(j) <= 0 {
					break drain
				}
			}
		}
		// Read and send the logged message, if there is one to read.
		i := C.get_message(j, &msg, &length)
		if i != -C.ENOENT && i != -C.EADDRNOTAVAIL {
			// Read the entry's timestamp.
			if C.sd_journal_get_realtime_usec(j, &stamp) != 0 {
				break
			}
			// Set up the time and text of the entry.
			timestamp := time.Unix(int64(stamp)/1000000, (int64(stamp)%1000000)*1000)
			line := append(C.GoBytes(unsafe.Pointer(msg), C.int(length)), "\n"...)
			// Recover the stream name by mapping
			// from the journal priority back to
			// the stream that we would have
			// assigned that value.
			source := ""
			if C.get_priority(j, &priority) != 0 {
				source = ""
			} else if priority == C.int(journal.PriErr) {
				source = "stderr"
			} else if priority == C.int(journal.PriInfo) {
				source = "stdout"
			}
			// Send the log message.
			cid := s.vars["CONTAINER_ID_FULL"]
			logWatcher.Msg <- &logger.Message{ContainerID: cid, Line: line, Source: source, Timestamp: timestamp}
		}
		// If we're at the end of the journal, we're done (for now).
		if C.sd_journal_next(j) <= 0 {
			break
		}
	}
	retCursor := ""
	if C.sd_journal_get_cursor(j, &cursor) == 0 {
		retCursor = C.GoString(cursor)
		C.free(unsafe.Pointer(cursor))
	}
	return retCursor
}

func (s *journald) followJournal(logWatcher *logger.LogWatcher, config logger.ReadConfig, j *C.sd_journal, pfd [2]C.int, cursor string) {
	s.readers.mu.Lock()
	s.readers.readers[logWatcher] = logWatcher
	s.readers.mu.Unlock()
	go func() {
		// Keep copying journal data out until we're notified to stop
		// or we hit an error.
		status := C.wait_for_data_or_close(j, pfd[0])
		for status == 1 {
			cursor = s.drainJournal(logWatcher, config, j, cursor)
			status = C.wait_for_data_or_close(j, pfd[0])
		}
		if status < 0 {
			cerrstr := C.strerror(C.int(-status))
			errstr := C.GoString(cerrstr)
			fmtstr := "error %q while attempting to follow journal for container %q"
			logrus.Errorf(fmtstr, errstr, s.vars["CONTAINER_ID_FULL"])
		}
		// Clean up.
		C.close(pfd[0])
		s.readers.mu.Lock()
		delete(s.readers.readers, logWatcher)
		s.readers.mu.Unlock()
		C.sd_journal_close(j)
		close(logWatcher.Msg)
	}()
	// Wait until we're told to stop.
	select {
	case <-logWatcher.WatchClose():
		// Notify the other goroutine that its work is done.
		C.close(pfd[1])
	}
}

func (s *journald) readLogs(logWatcher *logger.LogWatcher, config logger.ReadConfig) {
	var j *C.sd_journal
	var cmatch *C.char
	var stamp C.uint64_t
	var sinceUnixMicro uint64
	var pipes [2]C.int
	cursor := ""

{'fraction_non_alphanumeric': 0.1834625322997416, 'fraction_numerical': 0.018978882651697407, 'mean_word_length': 2.7880526493418833, 'pattern_counts': {'":': 0, '<': 36, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1053153', 'n_tokens_mistral': 5036, 'n_tokens_neox': 4577, 'n_words': 1102}