joerg/event
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

add journald

Browse source
This commit is contained in:
Jörg Prante 2024-06-28 13:38:59 +02:00
parent bef220d188
commit 710733f00d
10 changed files with 940 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
event-journald/build.gradle Normal file
View file

@ -0,0 +1,3 @@
dependencies {
implementation libs.jna
}

421
event-journald/src/main/java/org/xbib/event/journald/DefaultJournalEntry.java Normal file
View file

@ -0,0 +1,421 @@
package org.xbib.event.journald;
public class DefaultJournalEntry implements JournalEntry {
private String message;
private String messageId;
private int priority;
private String codeFile;
private String codeLine;
private String codeFunc;
private String errno;
private String syslogFacility;
private String syslogIdentifier;
private String syslogPid;
private String syslogTimestamp;
private String syslogRaw;
private String pid;
private String uid;
private String gid;
private String comm;
private String exe;
private String cmdLine;
private String capEffective;
private String auditSession;
private String auditLoginUid;
private String systemdCgroup;
private String systemdSlice;
private String systemdUnit;
private String systemdUserSlice;
private String systemdUserUnit;
private String systemdSession;
private String systemdOwnerUid;
private String selinuxContext;
private String sourceRealtimeTimestamp;
private String bootId;
private String machineId;
private String systemdInvocationId;
private String hostname;
private String transport;
private String streamId;
private String lineBreak;
public void setMessage(String message) {
this.message = message;
}
@Override
public String getMessage() {
return message;
}
public void setMessageId(String messageId) {
this.messageId = messageId;
}
@Override
public String getMessageId() {
return messageId;
}
public void setPriority(int priority) {
this.priority = priority;
}
@Override
public int getPriority() {
return priority;
}
public void setCodeFile(String codeFile) {
this.codeFile = codeFile;
}
@Override
public String getCodeFile() {
return codeFile;
}
public void setCodeLine(String codeLine) {
this.codeLine = codeLine;
}
@Override
public String getCodeLine() {
return codeLine;
}
public void setCodeFunc(String codeFunc) {
this.codeFunc = codeFunc;
}
@Override
public String getCodeFunc() {
return codeFunc;
}
public void setErrno(String errno) {
this.errno = errno;
}
@Override
public String getErrno() {
return errno;
}
public void setSyslogFacility(String syslogFacility) {
this.syslogFacility = syslogFacility;
}
@Override
public String getSyslogFacility() {
return syslogFacility;
}
public void setSyslogIdentifier(String syslogIdentifier) {
this.syslogIdentifier = syslogIdentifier;
}
@Override
public String getSyslogIdentifier() {
return syslogIdentifier;
}
public void setSyslogPid(String syslogPid) {
this.syslogPid = syslogPid;
}
@Override
public String getSyslogPid() {
return syslogPid;
}
public void setSyslogTimestamp(String syslogTimestamp) {
this.syslogTimestamp = syslogTimestamp;
}
@Override
public String getSyslogTimestamp() {
return syslogTimestamp;
}
public void setSyslogRaw(String syslogRaw) {
this.syslogRaw = syslogRaw;
}
@Override
public String getSyslogRaw() {
return syslogRaw;
}
public void setPid(String pid) {
this.pid = pid;
}
@Override
public String getPid() {
return pid;
}
public void setUid(String uid) {
this.uid = uid;
}
@Override
public String getUid() {
return uid;
}
public void setGid(String gid) {
this.gid = gid;
}
@Override
public String getGid() {
return gid;
}
public void setComm(String comm) {
this.comm = comm;
}
@Override
public String getComm() {
return comm;
}
public void setExe(String exe) {
this.exe = exe;
}
@Override
public String getExe() {
return exe;
}
public void setCmdLine(String cmdline) {
this.cmdLine = cmdline;
}
@Override
public String getCmdLine() {
return cmdLine;
}
public void setCapEffective(String capEffective) {
this.capEffective = capEffective;
}
@Override
public String getCapEffective() {
return capEffective;
}
public void setAuditSession(String auditSession) {
this.auditSession = auditSession;
}
@Override
public String getAuditSession() {
return auditSession;
}
public void setAuditLoginUid(String auditLoginUid) {
this.auditLoginUid = auditLoginUid;
}
@Override
public String getAuditLoginUid() {
return auditLoginUid;
}
public void setSystemdCgroup(String systemdCgroup) {
this.systemdCgroup = systemdCgroup;
}
@Override
public String getSystemdCgroup() {
return systemdCgroup;
}
public void setSystemdSlice(String systemdSlice) {
this.systemdSlice = systemdSlice;
}
@Override
public String getSystemdSlice() {
return systemdSlice;
}
public void setSystemdUnit(String systemdUnit) {
this.systemdUnit = systemdUnit;
}
@Override
public String getSystemdUnit() {
return systemdUnit;
}
public void setSystemdUserSlice(String systemdUserSlice) {
this.systemdUserSlice = systemdUserSlice;
}
@Override
public String getSystemdUserSlice() {
return systemdUserSlice;
}
public void setSystemdUserUnit(String systemdUserUnit) {
this.systemdUserUnit = systemdUserUnit;
}
@Override
public String getSystemdUserUnit() {
return systemdUserUnit;
}
public void setSystemdSession(String systemdSession) {
this.systemdSession = systemdSession;
}
@Override
public String getSystemdSession() {
return systemdSession;
}
public void setSystemdOwnerUid(String systemdOwnerUid) {
this.systemdOwnerUid = systemdOwnerUid;
}
@Override
public String getSystemdOwnerUid() {
return systemdOwnerUid;
}
public void setSelinuxContext(String selinuxContext) {
this.selinuxContext = selinuxContext;
}
@Override
public String getSelinuxContext() {
return selinuxContext;
}
public void setSourceRealtimeTimestamp(String sourceRealtimeTimestamp) {
this.sourceRealtimeTimestamp = sourceRealtimeTimestamp;
}
@Override
public String getSourceRealtimeTimestamp() {
return sourceRealtimeTimestamp;
}
public void setBootId(String bootId) {
this.bootId = bootId;
}
@Override
public String getBootId() {
return bootId;
}
public void setMachineId(String machineId) {
this.machineId = machineId;
}
@Override
public String getMachineId() {
return machineId;
}
public void setSystemdInvocationId(String systemdInvocationId) {
this.systemdInvocationId = systemdInvocationId;
}
@Override
public String getSystemdInvocationId() {
return systemdInvocationId;
}
public void setHostname(String hostname) {
this.hostname = hostname;
}
@Override
public String getHostname() {
return hostname;
}
public void setTransport(String transport) {
this.transport = transport;
}
@Override
public String getTransport() {
return transport;
}
public void setStreamId(String streamId) {
this.streamId = streamId;
}
@Override
public String getStreamId() {
return streamId;
}
public void setLineBreak(String lineBreak) {
this.lineBreak = lineBreak;
}
@Override
public String getLineBreak() {
return lineBreak;
}
@Override
public String getFieldValue(String fieldName) {
return null;
}
@Override
public String toString() {
return "priority=" + this.priority + ",message=" + this.message;
}
}

80
event-journald/src/main/java/org/xbib/event/journald/JournalEntry.java Normal file
View file

@ -0,0 +1,80 @@
package org.xbib.event.journald;
public interface JournalEntry {
String getMessage();
String getMessageId();
int getPriority();
String getCodeFile();
String getCodeLine();
String getCodeFunc();
String getErrno();
String getSyslogFacility();
String getSyslogIdentifier();
String getSyslogPid();
String getSyslogTimestamp();
String getSyslogRaw();
String getPid();
String getUid();
String getGid();
String getComm();
String getExe();
String getCmdLine();
String getCapEffective();
String getAuditSession();
String getAuditLoginUid();
String getSystemdCgroup();
String getSystemdSlice();
String getSystemdUnit();
String getSystemdUserSlice();
String getSystemdUserUnit();
String getSystemdSession();
String getSystemdOwnerUid();
String getSelinuxContext();
String getSourceRealtimeTimestamp();
String getBootId();
String getMachineId();
String getSystemdInvocationId();
String getHostname();
String getTransport();
String getStreamId();
String getLineBreak();
String getFieldValue(String fieldName);
}

240
event-journald/src/main/java/org/xbib/event/journald/SystemdJournalConsumer.java Normal file
View file

@ -0,0 +1,240 @@
package org.xbib.event.journald;
import com.sun.jna.Memory;
import com.sun.jna.Native;
import com.sun.jna.Pointer;
import com.sun.jna.StringArray;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;
import static org.xbib.event.journald.SystemdLibraryInstance.SD_JOURNAL_LOCAL_ONLY;
import static org.xbib.event.journald.SystemdLibraryInstance.SD_JOURNAL_NOP;
import static org.xbib.event.journald.SystemdLibraryInstance.SD_JOURNAL_SYSTEM;
public class SystemdJournalConsumer implements Runnable {
private static final Logger logger = Logger.getLogger(SystemdJournalConsumer.class.getName());
private final String match;
private final String field;
private final SystemdJournalListener listener;
public SystemdJournalConsumer(String match, SystemdJournalListener listener) {
this(match, null, listener);
}
public SystemdJournalConsumer(String match, String field, SystemdJournalListener listener) {
this.match = match;
this.field = field;
this.listener = listener;
}
@Override
public void run() {
try {
loop();
} catch (Throwable e) {
logger.log(Level.SEVERE, e.getMessage(), e);
}
}
/**
* <a href="https://gist.github.com/portante/ff7fb429c6f973aab377f7bb77b0ffdb">...</a>
* @throws IOException if loop fails
*/
private void loop() throws IOException {
SystemdLibraryInstance api = SystemdLibraryInstance.getInstance();
SystemdLibrary.SdJournal sdJournal = new SystemdLibrary.SdJournal();
int rc = api.sd_journal_open(sdJournal, SD_JOURNAL_LOCAL_ONLY);
if (rc < 0) {
throw new IOException("error opening journal for read: " + rc);
}
logger.log(Level.INFO, "top fd = " + sdJournal.top_level_fd +
" path = " + sdJournal.path +
" prefix = " + sdJournal.prefix +
" namespace = " + sdJournal.namespace);
if (match != null) {
rc = api.sd_journal_add_match(sdJournal, match, match.length());
logger.log(Level.INFO, "add_match: " + rc);
if (rc < 0) {
throw new IOException("error in add_match: " + rc);
}
}
rc = api.sd_journal_get_fd(sdJournal);
logger.log(Level.INFO, "get_fd: " + rc);
if (rc < 0) {
throw new IOException("failed to get file descriptor: " + rc);
}
rc = api.sd_journal_seek_tail(sdJournal);
logger.log(Level.INFO, "seek_tail: " + rc);
if (rc < 0) {
throw new IOException("failed to seek to tail of journal");
}
rc = api.sd_journal_previous(sdJournal);
logger.log(Level.INFO, "previous: " + rc);
rc = api.sd_journal_next(sdJournal);
logger.log(Level.INFO, "next: " + rc);
String[] strings = new String[1];
StringArray cursor = new StringArray(strings);
rc = api.sd_journal_get_cursor(sdJournal, cursor);
logger.log(Level.INFO, "get_cursor: " + rc);
System.exit(1);
while (true) {
do {
rc = api.sd_journal_wait(sdJournal, -1L);
logger.log(Level.INFO, "wait rc = " + rc);
} while (rc == SD_JOURNAL_NOP);
while ((rc = api.sd_journal_next(sdJournal)) > 0) {
logger.log(Level.INFO, "next: " + rc);
if (field != null) {
Pointer dataPointer = new Memory(Native.POINTER_SIZE);
Pointer sizePointer = new Memory(Native.POINTER_SIZE);
rc = api.sd_journal_get_data(sdJournal, field, dataPointer, sizePointer);
logger.log(Level.INFO, "get_data: " + rc);
if (rc != 0) {
throw new IOException("error in get_data: " + rc);
}
int size = sizePointer.getInt(0);
byte[] b = dataPointer.getByteArray(0, size);
String s = new String(b, StandardCharsets.UTF_8);
if (listener != null) {
listener.handleEntry(makeEntry(Collections.singletonList(s)));
}
} else {
String[] strings2 = new String[1];
StringArray nextCursor = new StringArray(strings2);
rc = api.sd_journal_get_cursor(sdJournal, nextCursor);
logger.log(Level.INFO, "get_cursor: " + rc);
if (!cursor.getString(0).equals(nextCursor.getString(0))) {
cursor = nextCursor;
Pointer dataPointer = new Memory(Native.POINTER_SIZE);
Pointer sizePointer = new Memory(Native.POINTER_SIZE);
List<String> list = new ArrayList<>();
while ((rc = api.sd_journal_enumerate_data(sdJournal, dataPointer, sizePointer)) > 0) {
logger.log(Level.INFO, "enumerate_data: " + rc);
int size = sizePointer.getInt(0);
byte[] b = dataPointer.getByteArray(0, size);
String s = new String(b, StandardCharsets.UTF_8);
list.add(s);
}
rc = api.sd_journal_restart_data(sdJournal);
logger.log(Level.INFO, "restart_data: " + rc);
if (listener != null) {
listener.handleEntry(makeEntry(list));
}
}
}
}
}
}
private JournalEntry makeEntry(List<String> list) {
DefaultJournalEntry journalEntry = new DefaultJournalEntry();
for (String string : list) {
if (string.startsWith("MESSAGE=")) {
journalEntry.setMessage(string.substring(8));
continue;
}
if (string.startsWith("MESSAGE_ID=")) {
journalEntry.setMessageId(string.substring(11));
continue;
}
if (string.startsWith("PRIORITY=")) {
journalEntry.setPriority(Integer.parseInt(string.substring(9)));
continue;
}
if (string.startsWith("CODE_FILE=")) {
journalEntry.setCodeFile(string.substring(10));
continue;
}
if (string.startsWith("CODE_LINE=")) {
journalEntry.setCodeLine(string.substring(10));
continue;
}
if (string.startsWith("CODE_FUNC=")) {
journalEntry.setCodeFunc(string.substring(10));
continue;
}
if (string.startsWith("ERRNO=")) {
journalEntry.setErrno(string.substring(6));
continue;
}
if (string.startsWith("SYSLOG_FACILITY=")) {
journalEntry.setSyslogFacility(string.substring(16));
continue;
}
if (string.startsWith("SYSLOG_IDENTIFIER=")) {
journalEntry.setSyslogIdentifier(string.substring(18));
continue;
}
if (string.startsWith("SYSLOG_PID=")) {
journalEntry.setSyslogPid(string.substring(11));
continue;
}
if (string.startsWith("SYSLOG_TIMESTAMP=")) {
journalEntry.setSyslogTimestamp(string.substring(17));
continue;
}
if (string.startsWith("SYSLOG_RAW=")) {
journalEntry.setSyslogRaw(string.substring(11));
continue;
}
if (string.startsWith("_PID=")) {
journalEntry.setPid(string.substring(5));
continue;
}
if (string.startsWith("_UID=")) {
journalEntry.setUid(string.substring(5));
continue;
}
if (string.startsWith("_GID=")) {
journalEntry.setGid(string.substring(5));
continue;
}
if (string.startsWith("_COMM=")) {
journalEntry.setComm(string.substring(6));
continue;
}
if (string.startsWith("_EXE=")) {
journalEntry.setExe(string.substring(5));
continue;
}
if (string.startsWith("_CMDLINE=")) {
journalEntry.setCmdLine(string.substring(9));
continue;
}
if (string.startsWith("_CAP_EFFECTIVE=")) {
journalEntry.setCapEffective(string.substring(15));
continue;
}
if (string.startsWith("_TRANSPORT=")) {
journalEntry.setTransport(string.substring(11));
continue;
}
if (string.startsWith("_SYSTEMD_OWNER_UID=")) {
journalEntry.setSystemdOwnerUid(string.substring(19));
continue;
}
if (string.startsWith("_SYSTEMD_UNIT=")) {
journalEntry.setSystemdUnit(string.substring(13));
continue;
}
if (string.startsWith("_SYSTEMD_USER_SLICE=")) {
journalEntry.setSystemdUserSlice(string.substring(19));
continue;
}
if (string.startsWith("_SYSTEMD_USER_UNIT=")) {
journalEntry.setSystemdUserUnit(string.substring(18));
}
}
return journalEntry;
}
}

8
event-journald/src/main/java/org/xbib/event/journald/SystemdJournalListener.java Normal file
View file

@ -0,0 +1,8 @@
package org.xbib.event.journald;
import java.io.IOException;
public interface SystemdJournalListener {
void handleEntry(JournalEntry entry) throws IOException;
}

49
event-journald/src/main/java/org/xbib/event/journald/SystemdLibrary.java Normal file
View file

@ -0,0 +1,49 @@
package org.xbib.event.journald;
import com.sun.jna.Library;
import com.sun.jna.Pointer;
import com.sun.jna.StringArray;
import com.sun.jna.Structure;
import java.util.Arrays;
import java.util.List;
public interface SystemdLibrary extends Library {
int sd_notify(int fd, String msg);
int sd_journal_send(String format, Object... args);
int sd_journal_open(SdJournal sdJournal, int flag);
int sd_journal_get_fd(SdJournal sdJournal);
int sd_journal_seek_tail(SdJournal sdJournal);
int sd_journal_previous(SdJournal sdJournal);
int sd_journal_next(SdJournal sdJournal);
int sd_journal_get_cursor(SdJournal sdJournal, StringArray cursor);
int sd_journal_add_match(SdJournal sdJournal, String match, int len);
int sd_journal_wait(SdJournal sdJournal, long timeout);
int sd_journal_get_data(SdJournal sdJournal, String field, Pointer data, Pointer length);
int sd_journal_enumerate_data(SdJournal sdJournal, Pointer data, Pointer length);
int sd_journal_restart_data(SdJournal sdJournal);
class SdJournal extends Structure {
public int top_level_fd;
public String path;
public String prefix;
public String namespace;
@Override
protected List<String> getFieldOrder() {
return Arrays.asList("top_level_fd","path","prefix","namespace");
}
}
}

105
event-journald/src/main/java/org/xbib/event/journald/SystemdLibraryInstance.java Normal file
View file

@ -0,0 +1,105 @@
package org.xbib.event.journald;
import com.sun.jna.Native;
import com.sun.jna.Pointer;
import com.sun.jna.StringArray;
import java.util.List;
/**
* The systemd library API, loaded by Java Native Access (JNA).
*
* The native library is loaded only once, so this class is a singleton.
*/
public class SystemdLibraryInstance {
private static final SystemdLibraryInstance instance = new SystemdLibraryInstance();
private final SystemdLibrary systemdLibrary;
private SystemdLibraryInstance() {
this.systemdLibrary = loadLibrary();
}
public static SystemdLibraryInstance getInstance() {
return instance;
}
public static final int SD_JOURNAL_LOCAL_ONLY = 1;
public static final int SD_JOURNAL_RUNTIME_ONLY = 2;
public static final int SD_JOURNAL_SYSTEM = 4;
public static final int SD_JOURNAL_CURRENT_USER = 8;
public static final int SD_JOURNAL_NOP = 0;
public static final int SD_JOURNAL_APPEND = 1;
public static final int SD_JOURNAL_INVALIDATE = 2;
public static final String SD_ID128_FORMAT_STR = "%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x";
public int sd_journal_send(String format, Object... args) {
return systemdLibrary.sd_journal_send(format, args);
}
public int sd_journal_send(String format, List<Object> args) {
return systemdLibrary.sd_journal_send(format, args.toArray());
}
public int sd_journal_open(SystemdLibrary.SdJournal sdJournal, int flag) {
return systemdLibrary.sd_journal_open(sdJournal, flag);
}
public int sd_journal_get_fd(SystemdLibrary.SdJournal sdJournal) {
return systemdLibrary.sd_journal_get_fd(sdJournal);
}
public int sd_journal_seek_tail(SystemdLibrary.SdJournal sdJournal) {
return systemdLibrary.sd_journal_seek_tail(sdJournal);
}
public int sd_journal_previous(SystemdLibrary.SdJournal sdJournal) {
return systemdLibrary.sd_journal_previous(sdJournal);
}
public int sd_journal_next(SystemdLibrary.SdJournal sdJournal) {
return systemdLibrary.sd_journal_next(sdJournal);
}
public int sd_journal_get_cursor(SystemdLibrary.SdJournal sdJournal, StringArray cursor) {
return systemdLibrary.sd_journal_get_cursor(sdJournal, cursor);
}
public int sd_journal_add_match(SystemdLibrary.SdJournal sdJournal, String match, int len) {
return systemdLibrary.sd_journal_add_match(sdJournal, match, len);
}
public int sd_journal_wait(SystemdLibrary.SdJournal sdJournal, long timeout_musec) {
return systemdLibrary.sd_journal_wait(sdJournal, timeout_musec);
}
public int sd_journal_get_data(SystemdLibrary.SdJournal sdJournal, String field, Pointer data, Pointer length) {
return systemdLibrary.sd_journal_get_data(sdJournal, field, data, length);
}
public int sd_journal_enumerate_data(SystemdLibrary.SdJournal sdJournal, Pointer data, Pointer length) {
return systemdLibrary.sd_journal_enumerate_data(sdJournal, data, length);
}
public int sd_journal_restart_data(SystemdLibrary.SdJournal sdJournal) {
return systemdLibrary.sd_journal_restart_data(sdJournal);
}
private static SystemdLibrary loadLibrary() {
try {
return Native.load("systemd", SystemdLibrary.class);
} catch (UnsatisfiedLinkError e) {
throw new RuntimeException("Failed to load systemd library." +
" Please note that JNA requires an executable temporary folder." +
" It can be explicitly defined with -Djna.tmpdir", e);
}
}
}

26
event-journald/src/test/java/org/xbib/event/journald/test/SystemdJournalReaderTest.java Normal file
View file

@ -0,0 +1,26 @@
package org.xbib.event.journald.test;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.EnabledOnOs;
import org.junit.jupiter.api.condition.OS;
import java.util.concurrent.Executors;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.xbib.event.journald.SystemdJournalConsumer;
@EnabledOnOs({OS.LINUX})
public class SystemdJournalReaderTest {
private static final Logger logger = Logger.getLogger(SystemdJournalReaderTest.class.getName());
@Test
void testConsumer() throws InterruptedException {
SystemdJournalConsumer consumer = new SystemdJournalConsumer(null,
entry -> logger.log(Level.INFO, entry.toString()));
Executors.newSingleThreadExecutor().submit(consumer);
// consuming for some seconds
Thread.sleep(60000L);
}
}

6
event-journald/src/test/resources/logging.properties Normal file
View file

@ -0,0 +1,6 @@
handlers=java.util.logging.ConsoleHandler
.level=ALL
java.util.logging.SimpleFormatter.format=%1$tY-%1$tm-%1$td %1$tH:%1$tM:%1$tS.%1$tL %4$-7s [%3$s] %5$s %6$s%n
java.util.logging.ConsoleHandler.level=ALL
java.util.logging.ConsoleHandler.formatter=java.util.logging.SimpleFormatter
jdk.event.security.level=INFO

2
settings.gradle
View file

@ -29,6 +29,7 @@ dependencyResolutionManagement {
library('net-http-client-netty-secure', 'org.xbib', 'net-http-client-netty-secure').versionRef('net-http') library('net-http-client-netty-secure', 'org.xbib', 'net-http-client-netty-secure').versionRef('net-http')
library('reactivestreams', 'org.reactivestreams', 'reactive-streams').version('1.0.3') library('reactivestreams', 'org.reactivestreams', 'reactive-streams').version('1.0.3')
library('rxjava3', 'io.reactivex.rxjava3', 'rxjava').version('3.0.3') library('rxjava3', 'io.reactivex.rxjava3', 'rxjava').version('3.0.3')
library('jna', 'net.java.dev.jna', 'jna').version('5.14.0')
} }
testLibs { testLibs {
version('junit', '5.10.2') version('junit', '5.10.2')
@ -44,6 +45,7 @@ dependencyResolutionManagement {
include 'event-api' include 'event-api'
include 'event-async' include 'event-async'
include 'event-common' include 'event-common'
include 'event-journald'
include 'event-loop' include 'event-loop'
include 'event-net-http' include 'event-net-http'
include 'event-pubsub' include 'event-pubsub'