bf94206849
continuous-integration/drone/push Build is passing
- Add POSTROUTING MASQUERADE rule alongside DNAT rules to ensure return traffic from container targets can route back through the same interface, matching legacy shell script behavior - Enhance FileWatcher to trigger periodic state reconciliation every tick regardless of config file changes, ensuring desired state is maintained after container restarts or iptables flushes
101 lines
2.7 KiB
Go
101 lines
2.7 KiB
Go
package watcher
|
|
|
|
import (
|
|
"crypto/md5"
|
|
"fmt"
|
|
"os"
|
|
"time"
|
|
|
|
"firewall_containers/network-go/logger"
|
|
)
|
|
|
|
// FileWatcher periodically polls a file for changes AND triggers a periodic
|
|
// reconciliation callback regardless of file changes.
|
|
type FileWatcher struct {
|
|
path string
|
|
period time.Duration
|
|
lastHash string
|
|
onChange func()
|
|
stopCh chan struct{}
|
|
}
|
|
|
|
// NewFileWatcher creates a new file watcher that:
|
|
// 1. Polls the file for content changes at the given period
|
|
// 2. Triggers a reconciliation callback every period regardless of changes
|
|
// to ensure desired state is maintained (stateful reconciliation)
|
|
func NewFileWatcher(path string, period time.Duration, onChange func()) *FileWatcher {
|
|
return &FileWatcher{
|
|
path: path,
|
|
period: period,
|
|
lastHash: "",
|
|
onChange: onChange,
|
|
stopCh: make(chan struct{}),
|
|
}
|
|
}
|
|
|
|
// hashFile computes an MD5 hash of the file contents
|
|
func (fw *FileWatcher) hashFile() (string, error) {
|
|
data, err := os.ReadFile(fw.path)
|
|
if err != nil {
|
|
return "", fmt.Errorf("failed to read file %s: %w", fw.path, err)
|
|
}
|
|
return fmt.Sprintf("%x", md5.Sum(data)), nil
|
|
}
|
|
|
|
// Start begins polling the file for changes in a goroutine.
|
|
// Every period, it checks if the file changed AND triggers a full reconciliation
|
|
// to maintain the desired state (handles container restarts, iptables flushes, etc.)
|
|
func (fw *FileWatcher) Start() {
|
|
// Compute initial hash
|
|
hash, err := fw.hashFile()
|
|
if err != nil {
|
|
logger.Warn("WATCHER: initial hash computation failed for %s: %v", fw.path, err)
|
|
} else {
|
|
fw.lastHash = hash
|
|
}
|
|
|
|
go func() {
|
|
ticker := time.NewTicker(fw.period)
|
|
defer ticker.Stop()
|
|
|
|
logger.Info("WATCHER: started watching %s every %s (periodic reconciliation enabled)", fw.path, fw.period)
|
|
|
|
for {
|
|
select {
|
|
case <-fw.stopCh:
|
|
logger.Info("WATCHER: stopped watching %s", fw.path)
|
|
return
|
|
case <-ticker.C:
|
|
hash, err := fw.hashFile()
|
|
if err != nil {
|
|
logger.Warn("WATCHER: failed to hash %s: %v", fw.path, err)
|
|
continue
|
|
}
|
|
|
|
fileChanged := hash != fw.lastHash
|
|
if fileChanged {
|
|
logger.Info("WATCHER: detected change in %s", fw.path)
|
|
fw.lastHash = hash
|
|
}
|
|
|
|
// Trigger reconciliation every period to maintain state,
|
|
// even if the config file hasn't changed.
|
|
// This ensures container restarts, iptable flushes, etc.
|
|
// are corrected.
|
|
if fw.onChange != nil {
|
|
if fileChanged {
|
|
logger.Info("WATCHER: triggering reconciliation (config changed)")
|
|
} else {
|
|
logger.Debug("WATCHER: triggering periodic state reconciliation")
|
|
}
|
|
fw.onChange()
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
// Stop signals the watcher goroutine to stop
|
|
func (fw *FileWatcher) Stop() {
|
|
close(fw.stopCh)
|
|
} |