fix: load logs on startup (#299)
This commit is contained in:
jamesread
parent
1e774f0078
commit
b9f55ab508
|
|
@ -0,0 +1,16 @@
|
|||
#
|
||||
# Integration Test Config: Log Persistence
|
||||
#
|
||||
|
||||
listenAddressSingleHTTPFrontend: 0.0.0.0:1337
|
||||
|
||||
logLevel: "DEBUG"
|
||||
checkForUpdates: false
|
||||
|
||||
saveLogs:
|
||||
resultsDirectory: /tmp/olivetin-test-logs
|
||||
|
||||
actions:
|
||||
- title: Echo Test
|
||||
shell: echo "Hello from persisted log test"
|
||||
icon: test
|
||||
|
|
@ -0,0 +1,258 @@
|
|||
import { describe, it, before, after } from 'mocha'
|
||||
import { expect } from 'chai'
|
||||
import { By, Condition } from 'selenium-webdriver'
|
||||
import fs from 'fs'
|
||||
import path from 'path'
|
||||
import {
|
||||
getRootAndWait,
|
||||
getActionButtons,
|
||||
takeScreenshotOnFailure,
|
||||
} from '../../lib/elements.js'
|
||||
|
||||
describe('config: logPersistence', function () {
|
||||
const logsDir = '/tmp/olivetin-test-logs'
|
||||
let firstExecutionTrackingId = null
|
||||
|
||||
before(async function () {
|
||||
// Clean up any existing test logs
|
||||
if (fs.existsSync(logsDir)) {
|
||||
fs.rmSync(logsDir, { recursive: true, force: true })
|
||||
}
|
||||
fs.mkdirSync(logsDir, { recursive: true })
|
||||
|
||||
await runner.start('logPersistence')
|
||||
})
|
||||
|
||||
after(async () => {
|
||||
await runner.stop()
|
||||
|
||||
// Clean up test logs directory
|
||||
if (fs.existsSync(logsDir)) {
|
||||
fs.rmSync(logsDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
afterEach(function () {
|
||||
takeScreenshotOnFailure(this.currentTest, webdriver)
|
||||
})
|
||||
|
||||
it('Execute action and verify log is saved to disk', async function () {
|
||||
this.timeout(30000)
|
||||
await getRootAndWait()
|
||||
|
||||
// Get initial log file count
|
||||
const initialLogCount = fs.existsSync(logsDir)
|
||||
? fs.readdirSync(logsDir).filter(f => f.endsWith('.yaml')).length
|
||||
: 0
|
||||
|
||||
// Wait for action button to be available
|
||||
await webdriver.wait(
|
||||
new Condition('wait for Echo Test button', async () => {
|
||||
const buttons = await webdriver.findElements(By.css('.action-button button'))
|
||||
for (const btn of buttons) {
|
||||
const text = await btn.getText()
|
||||
if (text.includes('Echo Test')) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}),
|
||||
10000
|
||||
)
|
||||
|
||||
// Find and click the Echo Test button
|
||||
const buttons = await webdriver.findElements(By.css('.action-button button'))
|
||||
let echoButton = null
|
||||
for (const btn of buttons) {
|
||||
const text = await btn.getText()
|
||||
if (text.includes('Echo Test')) {
|
||||
echoButton = btn
|
||||
break
|
||||
}
|
||||
}
|
||||
expect(echoButton).to.not.be.null
|
||||
|
||||
// Click the button to execute the action
|
||||
await echoButton.click()
|
||||
|
||||
// Wait for the log file to be written to disk
|
||||
await webdriver.wait(
|
||||
new Condition('wait for log file to appear', async () => {
|
||||
if (!fs.existsSync(logsDir)) {
|
||||
return false
|
||||
}
|
||||
const logFiles = fs.readdirSync(logsDir).filter(f => f.endsWith('.yaml'))
|
||||
return logFiles.length > initialLogCount
|
||||
}),
|
||||
10000
|
||||
)
|
||||
|
||||
// Wait a bit more to ensure file is fully written
|
||||
await webdriver.sleep(1000)
|
||||
|
||||
// Get the newest log file
|
||||
const logFiles = fs.readdirSync(logsDir).filter(f => f.endsWith('.yaml'))
|
||||
expect(logFiles.length).to.be.greaterThan(initialLogCount, 'At least one new log file should be saved')
|
||||
|
||||
// Sort by modification time to get the newest
|
||||
const logFilesWithStats = logFiles.map(f => {
|
||||
const filePath = path.join(logsDir, f)
|
||||
return {
|
||||
name: f,
|
||||
path: filePath,
|
||||
mtime: fs.statSync(filePath).mtime
|
||||
}
|
||||
}).sort((a, b) => b.mtime - a.mtime)
|
||||
|
||||
const newestLogFile = logFilesWithStats[0]
|
||||
expect(newestLogFile).to.not.be.undefined
|
||||
|
||||
// Read the log file to extract the tracking ID
|
||||
const logFileContent = fs.readFileSync(newestLogFile.path, 'utf8')
|
||||
|
||||
// Verify the log file contains expected content (action title might be in different fields)
|
||||
expect(logFileContent.length).to.be.greaterThan(0, 'Log file should not be empty')
|
||||
|
||||
// Extract tracking ID from filename first (most reliable)
|
||||
// Filename format: <title>.<timestamp>.<trackingId>.yaml
|
||||
// Tracking IDs are UUIDs, so match UUID pattern at the end before .yaml
|
||||
let uuidMatch = newestLogFile.name.match(/([a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12})\.yaml$/)
|
||||
|
||||
if (uuidMatch) {
|
||||
firstExecutionTrackingId = uuidMatch[1]
|
||||
} else {
|
||||
// Fallback: split by dots and get the last part before .yaml
|
||||
const parts = newestLogFile.name.replace('.yaml', '').split('.')
|
||||
if (parts.length >= 3) {
|
||||
// The last part should be the tracking ID
|
||||
firstExecutionTrackingId = parts[parts.length - 1]
|
||||
}
|
||||
}
|
||||
|
||||
// If still not found, try to extract from YAML content
|
||||
// Try different possible field name variations
|
||||
if (!firstExecutionTrackingId) {
|
||||
const patterns = [
|
||||
/executionTrackingID:\s*([^\s\n]+)/i,
|
||||
/execution_tracking_id:\s*([^\s\n]+)/i,
|
||||
/ExecutionTrackingID:\s*([^\s\n]+)/,
|
||||
/executionTrackingId:\s*([^\s\n]+)/i,
|
||||
]
|
||||
|
||||
for (const pattern of patterns) {
|
||||
const match = logFileContent.match(pattern)
|
||||
if (match) {
|
||||
firstExecutionTrackingId = match[1].trim()
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
expect(firstExecutionTrackingId).to.not.be.null
|
||||
expect(firstExecutionTrackingId.length).to.be.greaterThan(0)
|
||||
|
||||
// Verify the log file name contains the tracking ID
|
||||
expect(newestLogFile.name).to.include(firstExecutionTrackingId)
|
||||
|
||||
// Verify the log file content contains the action (might be in actionTitle, actionConfigTitle, or title field)
|
||||
const hasActionReference = logFileContent.includes('Echo Test') ||
|
||||
logFileContent.includes('echo') ||
|
||||
logFileContent.includes('actionTitle') ||
|
||||
logFileContent.includes('actionConfigTitle')
|
||||
expect(hasActionReference).to.be.true
|
||||
})
|
||||
|
||||
it('Restart service and verify logs are loaded from disk', async function () {
|
||||
this.timeout(60000)
|
||||
|
||||
// Skip if first test didn't set the tracking ID
|
||||
if (!firstExecutionTrackingId) {
|
||||
this.skip()
|
||||
}
|
||||
|
||||
// Verify log file exists before restart
|
||||
const logFilesBeforeRestart = fs.readdirSync(logsDir).filter(f => f.endsWith('.yaml'))
|
||||
expect(logFilesBeforeRestart.length).to.be.greaterThan(0, 'Log file should exist before restart')
|
||||
|
||||
// Find the log file for this execution
|
||||
const matchingLogFileBefore = logFilesBeforeRestart.find(f => f.includes(firstExecutionTrackingId))
|
||||
expect(matchingLogFileBefore).to.not.be.undefined
|
||||
|
||||
// Stop the current service instance
|
||||
await runner.stop()
|
||||
|
||||
// Wait a moment to ensure the process has fully stopped
|
||||
await new Promise((resolve) => setTimeout(resolve, 2000))
|
||||
|
||||
// Verify log file still exists after stop (should not be deleted)
|
||||
const logFilesAfterStop = fs.readdirSync(logsDir).filter(f => f.endsWith('.yaml'))
|
||||
expect(logFilesAfterStop.length).to.be.greaterThan(0, 'Log file should still exist after service stop')
|
||||
|
||||
const matchingLogFileAfter = logFilesAfterStop.find(f => f.includes(firstExecutionTrackingId))
|
||||
expect(matchingLogFileAfter).to.not.be.undefined
|
||||
|
||||
// Start a new service instance (logs should be loaded from disk)
|
||||
await runner.start('logPersistence')
|
||||
|
||||
// Wait for the service to fully start and load logs
|
||||
await new Promise((resolve) => setTimeout(resolve, 3000))
|
||||
|
||||
await getRootAndWait()
|
||||
|
||||
// Navigate directly to the specific log entry (this verifies the log was loaded)
|
||||
await webdriver.get(runner.baseUrl() + 'logs/' + firstExecutionTrackingId)
|
||||
|
||||
// Wait for the log details page to load
|
||||
await webdriver.wait(
|
||||
new Condition('wait for log details to load', async () => {
|
||||
try {
|
||||
const body = await webdriver.findElement(By.tagName('body'))
|
||||
const text = await body.getText()
|
||||
// The log should contain the output from the echo command
|
||||
return text.includes('Hello from persisted log test') || text.includes(firstExecutionTrackingId)
|
||||
} catch (e) {
|
||||
return false
|
||||
}
|
||||
}),
|
||||
15000
|
||||
)
|
||||
|
||||
// Verify the log content is displayed
|
||||
const body = await webdriver.findElement(By.tagName('body'))
|
||||
const bodyText = await body.getText()
|
||||
|
||||
// The persisted log should be accessible and contain the expected output
|
||||
expect(bodyText).to.include('Hello from persisted log test')
|
||||
})
|
||||
|
||||
it('Verify log file still exists after restart', async function () {
|
||||
// Skip if first test didn't set the tracking ID
|
||||
if (!firstExecutionTrackingId) {
|
||||
this.skip()
|
||||
}
|
||||
|
||||
// Verify the log file still exists on disk
|
||||
const logFiles = fs.readdirSync(logsDir).filter(f => f.endsWith('.yaml'))
|
||||
expect(logFiles.length).to.be.greaterThan(0, 'Log files should still exist after restart')
|
||||
|
||||
// Find the log file for the first execution
|
||||
const matchingLogFile = logFiles.find(f => f.includes(firstExecutionTrackingId))
|
||||
expect(matchingLogFile).to.not.be.undefined
|
||||
expect(matchingLogFile).to.not.be.null
|
||||
|
||||
// Verify the log file content is still valid
|
||||
const logFilePath = path.join(logsDir, matchingLogFile)
|
||||
const logFileContent = fs.readFileSync(logFilePath, 'utf8')
|
||||
expect(logFileContent.length).to.be.greaterThan(0, 'Log file should not be empty')
|
||||
|
||||
// The filename contains the tracking ID, so verify that
|
||||
expect(matchingLogFile).to.include(firstExecutionTrackingId)
|
||||
|
||||
// Verify the file contains some expected content (action reference)
|
||||
const hasActionReference = logFileContent.includes('Echo Test') ||
|
||||
logFileContent.includes('echo') ||
|
||||
logFileContent.includes('actionTitle') ||
|
||||
logFileContent.includes('actionConfigTitle')
|
||||
expect(hasActionReference).to.be.true
|
||||
})
|
||||
})
|
||||
|
|
@ -0,0 +1,216 @@
|
|||
package executor
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
"gopkg.in/yaml.v3"
|
||||
)
|
||||
|
||||
// LoadLogsFromDisk loads persisted logs from YAML files on disk and restores them to the executor.
|
||||
// This should be called during startup if saveLogs is configured.
|
||||
func (e *Executor) LoadLogsFromDisk() {
|
||||
resultsDir := e.Cfg.SaveLogs.ResultsDirectory
|
||||
if resultsDir == "" {
|
||||
return
|
||||
}
|
||||
|
||||
entries, skippedCount := e.readLogDirectory(resultsDir)
|
||||
if entries == nil {
|
||||
return
|
||||
}
|
||||
|
||||
loadedLogs, skippedCount := e.parseLogFiles(resultsDir, entries, skippedCount)
|
||||
|
||||
sort.Slice(loadedLogs, func(i, j int) bool {
|
||||
return loadedLogs[i].DatetimeStarted.Before(loadedLogs[j].DatetimeStarted)
|
||||
})
|
||||
|
||||
skippedCount = e.restoreLogsToExecutor(loadedLogs, skippedCount)
|
||||
|
||||
log.WithFields(log.Fields{
|
||||
"loaded": len(loadedLogs),
|
||||
"skipped": skippedCount,
|
||||
}).Info("Finished loading persisted logs from disk")
|
||||
}
|
||||
|
||||
// readLogDirectory reads the log directory and returns entries, or nil if the directory doesn't exist or can't be read.
|
||||
func (e *Executor) readLogDirectory(resultsDir string) ([]os.DirEntry, int) {
|
||||
if _, err := os.Stat(resultsDir); os.IsNotExist(err) {
|
||||
log.WithFields(log.Fields{
|
||||
"directory": resultsDir,
|
||||
}).Debug("Logs directory does not exist, skipping log loading")
|
||||
return nil, 0
|
||||
}
|
||||
|
||||
log.WithFields(log.Fields{
|
||||
"directory": resultsDir,
|
||||
}).Info("Loading persisted logs from disk")
|
||||
|
||||
entries, err := os.ReadDir(resultsDir)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
"directory": resultsDir,
|
||||
"error": err,
|
||||
}).Warnf("Failed to read logs directory")
|
||||
return nil, 0
|
||||
}
|
||||
|
||||
return entries, 0
|
||||
}
|
||||
|
||||
// parseLogFiles parses YAML log files from the directory entries.
|
||||
func (e *Executor) parseLogFiles(resultsDir string, entries []os.DirEntry, skippedCount int) ([]*InternalLogEntry, int) {
|
||||
loadedLogs := make([]*InternalLogEntry, 0)
|
||||
|
||||
for _, entry := range entries {
|
||||
if !e.shouldProcessLogEntry(entry) {
|
||||
continue
|
||||
}
|
||||
|
||||
logEntry, newSkippedCount := e.processLogFileEntry(resultsDir, entry.Name())
|
||||
skippedCount += newSkippedCount
|
||||
if logEntry != nil {
|
||||
loadedLogs = append(loadedLogs, logEntry)
|
||||
}
|
||||
}
|
||||
|
||||
return loadedLogs, skippedCount
|
||||
}
|
||||
|
||||
// shouldProcessLogEntry checks if a directory entry should be processed as a log file.
|
||||
func (e *Executor) shouldProcessLogEntry(entry os.DirEntry) bool {
|
||||
return !entry.IsDir() && strings.HasSuffix(entry.Name(), ".yaml")
|
||||
}
|
||||
|
||||
// processLogFileEntry processes a single log file entry and returns the log entry or nil if it should be skipped.
|
||||
func (e *Executor) processLogFileEntry(resultsDir, filename string) (*InternalLogEntry, int) {
|
||||
logEntry, ok := e.loadLogFileFromPath(resultsDir, filename)
|
||||
if !ok {
|
||||
return nil, 1
|
||||
}
|
||||
|
||||
if logEntry.ExecutionTrackingID == "" {
|
||||
log.WithFields(log.Fields{
|
||||
"file": filepath.Join(resultsDir, filename),
|
||||
}).Warnf("Log file missing execution tracking ID, skipping")
|
||||
return nil, 1
|
||||
}
|
||||
|
||||
e.restoreBindingForLogEntry(logEntry, filepath.Join(resultsDir, filename))
|
||||
return logEntry, 0
|
||||
}
|
||||
|
||||
// loadLogFileFromPath loads and unmarshals a single log file.
|
||||
func (e *Executor) loadLogFileFromPath(resultsDir, filename string) (*InternalLogEntry, bool) {
|
||||
filepath := filepath.Join(resultsDir, filename)
|
||||
data, err := os.ReadFile(filepath)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
"file": filepath,
|
||||
"error": err,
|
||||
}).Warnf("Failed to read log file")
|
||||
return nil, false
|
||||
}
|
||||
|
||||
var logEntry InternalLogEntry
|
||||
if err := yaml.Unmarshal(data, &logEntry); err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
"file": filepath,
|
||||
"error": err,
|
||||
}).Warnf("Failed to unmarshal log file")
|
||||
return nil, false
|
||||
}
|
||||
|
||||
return &logEntry, true
|
||||
}
|
||||
|
||||
// restoreBindingForLogEntry attempts to restore the binding for a log entry if it's missing or invalid.
|
||||
func (e *Executor) restoreBindingForLogEntry(logEntry *InternalLogEntry, filepath string) {
|
||||
if e.hasValidBinding(logEntry) || logEntry.ActionConfigTitle == "" {
|
||||
return
|
||||
}
|
||||
|
||||
binding := e.findBindingByActionTitle(logEntry.ActionConfigTitle, logEntry.EntityPrefix)
|
||||
if binding != nil {
|
||||
logEntry.Binding = binding
|
||||
return
|
||||
}
|
||||
|
||||
e.logBindingNotFound(logEntry, filepath)
|
||||
logEntry.Binding = nil
|
||||
}
|
||||
|
||||
// hasValidBinding checks if a log entry has a valid binding.
|
||||
func (e *Executor) hasValidBinding(logEntry *InternalLogEntry) bool {
|
||||
return logEntry.Binding != nil && logEntry.Binding.Action != nil
|
||||
}
|
||||
|
||||
// logBindingNotFound logs a debug message when a binding cannot be found for a log entry.
|
||||
func (e *Executor) logBindingNotFound(logEntry *InternalLogEntry, filepath string) {
|
||||
log.WithFields(log.Fields{
|
||||
"file": filepath,
|
||||
"actionTitle": logEntry.ActionConfigTitle,
|
||||
"entityPrefix": logEntry.EntityPrefix,
|
||||
"trackingId": logEntry.ExecutionTrackingID,
|
||||
}).Debug("Could not find binding for log entry, loading without binding")
|
||||
}
|
||||
|
||||
// restoreLogsToExecutor restores loaded logs to the executor's internal structures.
|
||||
func (e *Executor) restoreLogsToExecutor(loadedLogs []*InternalLogEntry, skippedCount int) int {
|
||||
e.logmutex.Lock()
|
||||
defer e.logmutex.Unlock()
|
||||
|
||||
for _, logEntry := range loadedLogs {
|
||||
if _, exists := e.logs[logEntry.ExecutionTrackingID]; exists {
|
||||
log.WithFields(log.Fields{
|
||||
"trackingId": logEntry.ExecutionTrackingID,
|
||||
}).Debug("Log entry already exists, skipping")
|
||||
skippedCount++
|
||||
continue
|
||||
}
|
||||
|
||||
logEntry.Index = int64(len(e.logsTrackingIdsByDate))
|
||||
e.logs[logEntry.ExecutionTrackingID] = logEntry
|
||||
e.logsTrackingIdsByDate = append(e.logsTrackingIdsByDate, logEntry.ExecutionTrackingID)
|
||||
|
||||
if logEntry.Binding != nil {
|
||||
e.addLogToBindingMap(logEntry)
|
||||
}
|
||||
}
|
||||
|
||||
return skippedCount
|
||||
}
|
||||
|
||||
// addLogToBindingMap adds a log entry to the LogsByBindingId map.
|
||||
func (e *Executor) addLogToBindingMap(logEntry *InternalLogEntry) {
|
||||
if _, containsKey := e.LogsByBindingId[logEntry.Binding.ID]; !containsKey {
|
||||
e.LogsByBindingId[logEntry.Binding.ID] = make([]*InternalLogEntry, 0)
|
||||
}
|
||||
e.LogsByBindingId[logEntry.Binding.ID] = append(e.LogsByBindingId[logEntry.Binding.ID], logEntry)
|
||||
}
|
||||
|
||||
// findBindingByActionTitle attempts to find a binding by matching the action config title and entity prefix.
|
||||
func (e *Executor) findBindingByActionTitle(actionConfigTitle string, entityPrefix string) *ActionBinding {
|
||||
e.MapActionBindingsLock.RLock()
|
||||
defer e.MapActionBindingsLock.RUnlock()
|
||||
|
||||
for _, binding := range e.MapActionBindings {
|
||||
if binding.Action.Title == actionConfigTitle && e.matchesEntityPrefix(binding, entityPrefix) {
|
||||
return binding
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// matchesEntityPrefix checks if a binding matches the given entity prefix.
|
||||
func (e *Executor) matchesEntityPrefix(binding *ActionBinding, entityPrefix string) bool {
|
||||
if entityPrefix == "" {
|
||||
return binding.Entity == nil
|
||||
}
|
||||
return binding.Entity != nil && binding.Entity.UniqueKey == entityPrefix
|
||||
}
|
||||
|
|
@ -257,6 +257,8 @@ func main() {
|
|||
executor.RebuildActionMap()
|
||||
config.AddListener(executor.RebuildActionMap)
|
||||
|
||||
executor.LoadLogsFromDisk()
|
||||
|
||||
go onstartup.Execute(cfg, executor)
|
||||
go oncron.Schedule(cfg, executor)
|
||||
go onfileindir.WatchFilesInDirectory(cfg, executor)
|
||||
|
|
|
|||
Loading…
Reference in New Issue