Class H2LogStore

java.lang.Object

com.scivicslab.actoriac.log.H2LogStore

All Implemented Interfaces:

DistributedLogStore, AutoCloseable

public class H2LogStore
extends Object
implements DistributedLogStore

H2 Database implementation of DistributedLogStore.

Uses H2 embedded database for storing logs from distributed workflow execution. Supports concurrent access from multiple threads and provides efficient querying by node, level, and time.

Features:

• Pure Java - no native dependencies
• Single file storage (.mv.db)
• Asynchronous batch writing for performance
• SQL-based querying

Author:

devteam@scivics-lab.com

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
private static interface	H2LogStore.LogTask

Field Summary

Fields

Modifier and Type	Field	Description
private static final int	BATCH_SIZE
private final Connection	connection
private static final int	DEFAULT_TCP_PORT
private static final Logger	LOG
private final AtomicBoolean	running
private final BlockingQueue<H2LogStore.LogTask>	writeQueue
private final Thread	writerThread

Constructor Summary

Constructors

Constructor	Description
H2LogStore()	Creates an in-memory H2LogStore (for testing).
H2LogStore(String host, int port, String dbPath)	Creates an H2LogStore connected to a remote TCP server.
H2LogStore(Path dbPath)	Creates an H2LogStore with the specified database path.

Method Summary

Modifier and Type	Method	Description
void	close()
static H2LogStore	createWithFallback(String logServer, Path embeddedPath)	Factory method that attempts TCP connection with fallback to embedded mode.
void	endSession(long sessionId, SessionStatus status)	Ends a session with the given status.
private void	flushWrites()
long	getLatestSessionId()	Gets the most recent session ID.
List<LogEntry>	getLogsByLevel(long sessionId, LogLevel minLevel)	Retrieves log entries filtered by level.
List<LogEntry>	getLogsByNode(long sessionId, String nodeId)	Retrieves all log entries for a specific node in a session.
SessionSummary	getSummary(long sessionId)	Gets a summary of the session.
private void	initSchema()
List<SessionSummary>	listSessions(int limit)	Lists all sessions.
void	log(long sessionId, String nodeId, LogLevel level, String message)	Records a log entry.
void	log(long sessionId, String nodeId, String vertexName, LogLevel level, String message)	Records a log entry with vertex context.
void	logAction(long sessionId, String nodeId, String vertexName, String actionName, int exitCode, long durationMs, String output)	Records an action result.
private LogEntry	mapLogEntry(ResultSet rs)
void	markNodeFailed(long sessionId, String nodeId, String reason)	Marks a node as failed in this session.
void	markNodeSuccess(long sessionId, String nodeId)	Marks a node as succeeded in this session.
private void	processBatch(List<H2LogStore.LogTask> batch)
long	startSession(String workflowName, int nodeCount)	Starts a new workflow execution session.
long	startSession(String workflowName, String overlayName, String inventoryName, int nodeCount)	Starts a new workflow execution session with overlay and inventory info.
private void	verifySchema()	Verifies the database schema exists (for TCP connections).
private void	writerLoop()

Methods inherited from class Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface DistributedLogStore

getErrors

Field Details

LOG

private static final Logger LOG

connection

private final Connection connection

writeQueue

private final BlockingQueue<H2LogStore.LogTask> writeQueue

writerThread

private final Thread writerThread

running

private final AtomicBoolean running

BATCH_SIZE

private static final int BATCH_SIZE

See Also:

• Constant Field Values

DEFAULT_TCP_PORT

private static final int DEFAULT_TCP_PORT

See Also:

• Constant Field Values

Constructor Details

H2LogStore

public H2LogStore(Path dbPath)
           throws SQLException

Creates an H2LogStore with the specified database path.

Parameters:

dbPath - path to the database file (without extension)

Throws:

SQLException - if database connection fails

H2LogStore

public H2LogStore()
           throws SQLException

Creates an in-memory H2LogStore (for testing).

Throws:

SQLException - if database connection fails

H2LogStore

public H2LogStore(String host,
 int port,
 String dbPath)
           throws SQLException

Creates an H2LogStore connected to a remote TCP server.

The server should be started using the log-server command before connecting. Schema is expected to be initialized by the server.

Parameters:

host - H2 server hostname (typically "localhost")

port - H2 server TCP port

dbPath - database path on the server

Throws:

SQLException - if connection fails

Method Details

createWithFallback

public static H2LogStore createWithFallback(String logServer,
 Path embeddedPath)
                                     throws SQLException

Factory method that attempts TCP connection with fallback to embedded mode.

If a log server address is specified and reachable, connects via TCP. Otherwise, falls back to embedded mode with AUTO_SERVER=TRUE.

Parameters:

logServer - server address in "host:port" format (may be null)

embeddedPath - path for embedded database fallback

Returns:

H2LogStore instance

Throws:

SQLException - if both TCP and embedded connections fail

verifySchema

private void verifySchema()
                   throws SQLException

Verifies the database schema exists (for TCP connections).

Throws:

SQLException - if schema is not initialized

initSchema

private void initSchema()
                 throws SQLException

Throws:

SQLException

writerLoop

private void writerLoop()

processBatch

private void processBatch(List<H2LogStore.LogTask> batch)

startSession

public long startSession(String workflowName,
 int nodeCount)

Description copied from interface: DistributedLogStore

Starts a new workflow execution session.

Specified by:

startSession in interface DistributedLogStore

Parameters:

workflowName - name of the workflow being executed

nodeCount - number of nodes participating in this session

Returns:

session ID for subsequent log entries

startSession

public long startSession(String workflowName,
 String overlayName,
 String inventoryName,
 int nodeCount)

Description copied from interface: DistributedLogStore

Starts a new workflow execution session with overlay and inventory info.

Specified by:

startSession in interface DistributedLogStore

Parameters:

workflowName - name of the workflow being executed

overlayName - name of the overlay being used (may be null)

inventoryName - name of the inventory file being used (may be null)

nodeCount - number of nodes participating in this session

Returns:

session ID for subsequent log entries

log

public void log(long sessionId,
 String nodeId,
 LogLevel level,
 String message)

Description copied from interface: DistributedLogStore

Records a log entry.

Specified by:

log in interface DistributedLogStore

Parameters:

sessionId - session ID from startSession()

nodeId - identifier of the node generating this log

level - log level

message - log message

log

public void log(long sessionId,
 String nodeId,
 String vertexName,
 LogLevel level,
 String message)

Description copied from interface: DistributedLogStore

Records a log entry with vertex context.

Specified by:

log in interface DistributedLogStore

Parameters:

sessionId - session ID from startSession()

nodeId - identifier of the node

vertexName - current vertex name in workflow

level - log level

message - log message

logAction

public void logAction(long sessionId,
 String nodeId,
 String vertexName,
 String actionName,
 int exitCode,
 long durationMs,
 String output)

Description copied from interface: DistributedLogStore

Records an action result.

Specified by:

logAction in interface DistributedLogStore

Parameters:

sessionId - session ID

nodeId - node identifier

vertexName - vertex name

actionName - action/method name

exitCode - command exit code (0 for success)

durationMs - execution duration in milliseconds

output - command output or result message

markNodeSuccess

public void markNodeSuccess(long sessionId,
 String nodeId)

Description copied from interface: DistributedLogStore

Marks a node as succeeded in this session.

Specified by:

markNodeSuccess in interface DistributedLogStore

Parameters:

sessionId - session ID

nodeId - node identifier

markNodeFailed

public void markNodeFailed(long sessionId,
 String nodeId,
 String reason)

Description copied from interface: DistributedLogStore

Marks a node as failed in this session.

Specified by:

markNodeFailed in interface DistributedLogStore

Parameters:

sessionId - session ID

nodeId - node identifier

reason - failure reason

endSession

public void endSession(long sessionId,
 SessionStatus status)

Description copied from interface: DistributedLogStore

Ends a session with the given status.

Specified by:

endSession in interface DistributedLogStore

Parameters:

sessionId - session ID

status - final status

flushWrites

private void flushWrites()

getLogsByNode

public List<LogEntry> getLogsByNode(long sessionId,
 String nodeId)

Description copied from interface: DistributedLogStore

Retrieves all log entries for a specific node in a session.

Specified by:

getLogsByNode in interface DistributedLogStore

Parameters:

sessionId - session ID

nodeId - node identifier

Returns:

list of log entries

getLogsByLevel

public List<LogEntry> getLogsByLevel(long sessionId,
 LogLevel minLevel)

Description copied from interface: DistributedLogStore

Retrieves log entries filtered by level.

Specified by:

getLogsByLevel in interface DistributedLogStore

Parameters:

sessionId - session ID

minLevel - minimum log level to include

Returns:

list of log entries

getSummary

public SessionSummary getSummary(long sessionId)

Description copied from interface: DistributedLogStore

Gets a summary of the session.

Specified by:

getSummary in interface DistributedLogStore

Parameters:

sessionId - session ID

Returns:

session summary

getLatestSessionId

public long getLatestSessionId()

Description copied from interface: DistributedLogStore

Gets the most recent session ID.

Specified by:

getLatestSessionId in interface DistributedLogStore

Returns:

latest session ID, or -1 if no sessions exist

listSessions

public List<SessionSummary> listSessions(int limit)

Description copied from interface: DistributedLogStore

Lists all sessions.

Specified by:

listSessions in interface DistributedLogStore

Parameters:

limit - maximum number of sessions to return

Returns:

list of session summaries

mapLogEntry

private LogEntry mapLogEntry(ResultSet rs)
                      throws SQLException

Throws:

SQLException

close

public void close()
           throws Exception

Specified by:

close in interface AutoCloseable

Throws:

Exception