unkin-agent
415bf0cce1
Single Go binary for the ClickHouse log store (logs.raw): chlog with cat/tail/grep subcommands, plus chcat/chtail/chgrep argv[0]-dispatched symlink entrypoints. Every query is time-bounded and fully parameterized; chgrep guards wide unfiltered scans. Ships nfpm RPM with completions and woodpecker PR/tag pipelines mirroring node-lookup.
85 lines
1.6 KiB
Go
85 lines
1.6 KiB
Go
package chlog
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
)
|
|
|
|
const (
|
|
TailInterval = 2 * time.Second
|
|
tailOverlap = 5 * time.Second
|
|
)
|
|
|
|
// Tail streams the initial window then polls every interval. Each poll
|
|
// re-queries from a little before the last-seen timestamp and drops rows
|
|
// already emitted, so late-arriving rows inside the overlap still surface.
|
|
func Tail(ctx context.Context, c runner, f Filter, now func() time.Time, interval time.Duration, emit func(Row) error) error {
|
|
if now == nil {
|
|
now = time.Now
|
|
}
|
|
if interval <= 0 {
|
|
interval = TailInterval
|
|
}
|
|
|
|
seen := map[uint64]time.Time{}
|
|
lastTS := f.Since
|
|
track := func(r Row) error {
|
|
if ts := r.Time(); ts.After(lastTS) {
|
|
lastTS = ts
|
|
}
|
|
seen[r.Key()] = r.Time()
|
|
return emit(r)
|
|
}
|
|
|
|
first := f
|
|
first.Until = now().UTC()
|
|
first.Limit = 0
|
|
if _, err := Page(ctx, c, first, 0, track); err != nil {
|
|
if ctx.Err() != nil {
|
|
return ctx.Err()
|
|
}
|
|
return err
|
|
}
|
|
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-ticker.C:
|
|
}
|
|
|
|
since := lastTS.Add(-tailOverlap)
|
|
if since.Before(f.Since) {
|
|
since = f.Since
|
|
}
|
|
pf := f
|
|
pf.Since = since
|
|
pf.Until = now().UTC()
|
|
pf.Limit = 0
|
|
if !pf.Since.Before(pf.Until) {
|
|
continue
|
|
}
|
|
_, err := Page(ctx, c, pf, 0, func(r Row) error {
|
|
if _, dup := seen[r.Key()]; dup {
|
|
return nil
|
|
}
|
|
return track(r)
|
|
})
|
|
if err != nil {
|
|
if ctx.Err() != nil {
|
|
return ctx.Err()
|
|
}
|
|
return err
|
|
}
|
|
|
|
floor := lastTS.Add(-2 * tailOverlap)
|
|
for k, ts := range seen {
|
|
if ts.Before(floor) {
|
|
delete(seen, k)
|
|
}
|
|
}
|
|
}
|
|
}
|