feat(components/query): move parse time func, filter to common packages

components/accelerator/nvidia/fabric-manager/config.go

@@ -6,8 +6,8 @@ import (
 
 	fabric_manager_log "github.com/leptonai/gpud/components/accelerator/nvidia/query/fabric-manager-log"
 	query_config "github.com/leptonai/gpud/components/query/config"
+	query_log_common "github.com/leptonai/gpud/components/query/log/common"
 	query_log_config "github.com/leptonai/gpud/components/query/log/config"
-	query_log_filter "github.com/leptonai/gpud/components/query/log/filter"
 
 	"k8s.io/utils/ptr"
 )
@@ -54,7 +54,7 @@ const (
 )
 
 var (
-	filters = []*query_log_filter.Filter{
+	filters = []*query_log_common.Filter{
 		{
 			Name:            eventNVSwitchFatailSXid,
 			Regex:           ptr.To(fabric_manager_log.RegexNVSwitchFatalSXidFromLog),

components/diagnose/diagnose.go

@@ -11,7 +11,7 @@ import (
 
 	nvidia_query "github.com/leptonai/gpud/components/accelerator/nvidia/query"
 	"github.com/leptonai/gpud/components/dmesg"
-	query_log_filter "github.com/leptonai/gpud/components/query/log/filter"
+	query_log_common "github.com/leptonai/gpud/components/query/log/common"
 	query_log_tail "github.com/leptonai/gpud/components/query/log/tail"
 	"github.com/leptonai/gpud/pkg/host"
 	"github.com/leptonai/gpud/pkg/process"
@@ -149,7 +149,7 @@ func run(ctx context.Context, dir string, opts ...OpOption) error {
 		query_log_tail.WithLinesToTail(5000),
 		query_log_tail.WithSelectFilter(defaultDmesgCfg.Log.SelectFilters...),
 		query_log_tail.WithParseTime(dmesg.ExtractTimeFromLogLine),
-		query_log_tail.WithProcessMatched(func(line []byte, time time.Time, matched *query_log_filter.Filter) {
+		query_log_tail.WithProcessMatched(func(line []byte, time time.Time, matched *query_log_common.Filter) {
 			o.CheckSummary = append(o.CheckSummary, fmt.Sprintf("dmesg match: %s", string(line)))
 		}),
 	)

components/diagnose/scan.go

@@ -12,7 +12,7 @@ import (
 	nvidia_query_sxid "github.com/leptonai/gpud/components/accelerator/nvidia/query/sxid"
 	nvidia_query_xid "github.com/leptonai/gpud/components/accelerator/nvidia/query/xid"
 	"github.com/leptonai/gpud/components/dmesg"
-	query_log_filter "github.com/leptonai/gpud/components/query/log/filter"
+	query_log_common "github.com/leptonai/gpud/components/query/log/common"
 	query_log_tail "github.com/leptonai/gpud/components/query/log/tail"
 	"github.com/leptonai/gpud/log"
 	"github.com/leptonai/gpud/pkg/file"
@@ -184,7 +184,7 @@ func Scan(ctx context.Context, opts ...OpOption) error {
 		query_log_tail.WithLinesToTail(op.lines),
 		query_log_tail.WithSelectFilter(defaultDmesgCfg.Log.SelectFilters...),
 		query_log_tail.WithParseTime(dmesg.ExtractTimeFromLogLine),
-		query_log_tail.WithProcessMatched(func(line []byte, time time.Time, matched *query_log_filter.Filter) {
+		query_log_tail.WithProcessMatched(func(line []byte, time time.Time, matched *query_log_common.Filter) {
 			log.Logger.Debugw("matched", "line", string(line))
 			matchedB, _ := matched.YAML()
 			fmt.Println(string(matchedB))

components/dmesg/component_event.go

@@ -9,7 +9,7 @@ import (
 
 	"github.com/leptonai/gpud/components"
 	query_log "github.com/leptonai/gpud/components/query/log"
-	query_log_filter "github.com/leptonai/gpud/components/query/log/filter"
+	query_log_common "github.com/leptonai/gpud/components/query/log/common"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -49,9 +49,9 @@ func ParseEventDmesgMatched(m map[string]string) (query_log.Item, error) {
 	ev.Time = metav1.Time{Time: time.Unix(unixSeconds, 0)}
 	ev.Line = m[EventKeyDmesgMatchedLine]
 
-	var f *query_log_filter.Filter
+	var f *query_log_common.Filter
 	if m[EventKeyDmesgMatchedFilter] != "" {
-		f, err = query_log_filter.ParseFilterJSON([]byte(m[EventKeyDmesgMatchedFilter]))
+		f, err = query_log_common.ParseFilterJSON([]byte(m[EventKeyDmesgMatchedFilter]))
 		if err != nil {
 			return query_log.Item{}, err
 		}

components/dmesg/component_state.go

@@ -9,7 +9,7 @@ import (
 
 	"github.com/leptonai/gpud/components"
 	query_log "github.com/leptonai/gpud/components/query/log"
-	query_log_filter "github.com/leptonai/gpud/components/query/log/filter"
+	query_log_common "github.com/leptonai/gpud/components/query/log/common"
 
 	"github.com/nxadm/tail"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -78,10 +78,10 @@ func ParseStateDmesgTailScanMatched(m map[string]string) (query_log.Item, error)
 	}
 	ev.Line = m[StateKeyDmesgTailScanMatchedLine]
 
-	var f *query_log_filter.Filter
+	var f *query_log_common.Filter
 	if m[StateKeyDmesgTailScanMatchedFilter] != "" {
 		var err error
-		f, err = query_log_filter.ParseFilterJSON([]byte(m[StateKeyDmesgTailScanMatchedFilter]))
+		f, err = query_log_common.ParseFilterJSON([]byte(m[StateKeyDmesgTailScanMatchedFilter]))
 		if err != nil {
 			return query_log.Item{}, err
 		}

components/dmesg/component_test.go

@@ -7,8 +7,8 @@ import (
 	"time"
 
 	query_config "github.com/leptonai/gpud/components/query/config"
+	query_log_common "github.com/leptonai/gpud/components/query/log/common"
 	query_log_config "github.com/leptonai/gpud/components/query/log/config"
-	query_log_filter "github.com/leptonai/gpud/components/query/log/filter"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
@@ -31,7 +31,7 @@ func TestComponent(t *testing.T) {
 	defer cancel()
 
 	xidErr := "NVRM: Xid (0000:03:00): 14, Channel 00000001"
-	filters := []*query_log_filter.Filter{
+	filters := []*query_log_common.Filter{
 		{
 			Name:      "xid error check",
 			Substring: &xidErr,

components/dmesg/filters.go

@@ -5,7 +5,7 @@ import (
 
 	nvidia_query "github.com/leptonai/gpud/components/accelerator/nvidia/query"
 	"github.com/leptonai/gpud/components/memory"
-	query_log_filter "github.com/leptonai/gpud/components/query/log/filter"
+	query_log_common "github.com/leptonai/gpud/components/query/log/common"
 
 	"k8s.io/utils/ptr"
 )
@@ -32,8 +32,8 @@ const (
 	EventOOMCgroupRegex = `Memory cgroup out of memory`
 )
 
-func DefaultLogFilters(ctx context.Context) ([]*query_log_filter.Filter, error) {
-	defaultFilters := []*query_log_filter.Filter{
+func DefaultLogFilters(ctx context.Context) ([]*query_log_common.Filter, error) {
+	defaultFilters := []*query_log_common.Filter{
 		{
 			Name:            EventOOMKill,
 			Regex:           ptr.To(EventOOMKillRegex),

components/dmesg/filters_nvidia.go

@@ -8,7 +8,7 @@ import (
 	nvidia_query_peermem "github.com/leptonai/gpud/components/accelerator/nvidia/query/peermem"
 	nvidia_query_sxid "github.com/leptonai/gpud/components/accelerator/nvidia/query/sxid"
 	nvidia_query_xid "github.com/leptonai/gpud/components/accelerator/nvidia/query/xid"
-	query_log_filter "github.com/leptonai/gpud/components/query/log/filter"
+	query_log_common "github.com/leptonai/gpud/components/query/log/common"
 
 	"k8s.io/utils/ptr"
 )
@@ -45,8 +45,8 @@ const (
 	EventNvidiaNCCLSegfaultInLibnccl = "nvidia_nccl_segfault_in_libnccl"
 )
 
-func DefaultDmesgFiltersForNvidia() []*query_log_filter.Filter {
-	return []*query_log_filter.Filter{
+func DefaultDmesgFiltersForNvidia() []*query_log_common.Filter {
+	return []*query_log_common.Filter{
 		{
 			Name:            EventNvidiaNVRMXid,
 			Regex:           ptr.To(nvidia_query_xid.RegexNVRMXidDmesg),

components/query/log/common/common.go

@@ -0,0 +1,10 @@
+// Package common provides the common log components.
+package common
+
+import (
+	"time"
+)
+
+type ParseTimeFunc func([]byte) (time.Time, error)
+
+type ProcessMatchedFunc func(line []byte, parsedTime time.Time, filter *Filter)

components/query/log/filter/filter.go → components/query/log/common/filter.go

@@ -1,5 +1,4 @@
-// Package filter provides the log filter implementation.
-package filter
+package common
 
 import (
 	"bytes"

components/query/log/config/config.go

@@ -8,7 +8,7 @@ import (
 	"errors"
 
 	query_config "github.com/leptonai/gpud/components/query/config"
-	query_log_filter "github.com/leptonai/gpud/components/query/log/filter"
+	query_log_common "github.com/leptonai/gpud/components/query/log/common"
 
 	"github.com/nxadm/tail"
 )
@@ -32,12 +32,12 @@ type Config struct {
 	// An event is generated if any of the filters match.
 	// Useful for explicit blacklisting "error" logs
 	// (e.g., GPU error messages in dmesg).
-	SelectFilters []*query_log_filter.Filter `json:"select_filters"`
+	SelectFilters []*query_log_common.Filter `json:"select_filters"`
 	// "AND" conditions to select logs.
 	// An event is generated if all of the filters do not match.
 	// Useful for explicit whitelisting logs and catch all other
 	// (e.g., good healthy log messages).
-	RejectFilters []*query_log_filter.Filter `json:"reject_filters"`
+	RejectFilters []*query_log_common.Filter `json:"reject_filters"`
 
 	DB       *sql.DB        `json:"-"`
 	SeekInfo *tail.SeekInfo `json:"seek_info,omitempty"`

components/query/log/poller.go

@@ -7,8 +7,8 @@ import (
 	"time"
 
 	"github.com/leptonai/gpud/components/query"
+	query_log_common "github.com/leptonai/gpud/components/query/log/common"
 	query_log_config "github.com/leptonai/gpud/components/query/log/config"
-	query_log_filter "github.com/leptonai/gpud/components/query/log/filter"
 	query_log_tail "github.com/leptonai/gpud/components/query/log/tail"
 	"github.com/leptonai/gpud/log"
 
@@ -51,7 +51,7 @@ type Poller interface {
 	// Returns all the events for the given "since" time.
 	// If none, it returns all events that are already filtered
 	// by the default filters in the configuration.
-	Find(since time.Time, selectFilters ...*query_log_filter.Filter) ([]Item, error)
+	Find(since time.Time, selectFilters ...*query_log_common.Filter) ([]Item, error)
 
 	// Returns the last seek info.
 	SeekInfo() tail.SeekInfo
@@ -65,7 +65,7 @@ type Item struct {
 	Line string      `json:"line"`
 
 	// Matched filter that was applied to this item/line.
-	Matched *query_log_filter.Filter `json:"matched,omitempty"`
+	Matched *query_log_common.Filter `json:"matched,omitempty"`
 
 	Error error `json:"error,omitempty"`
 }
@@ -84,11 +84,11 @@ type poller struct {
 	bufferedItems   []Item
 }
 
-func New(ctx context.Context, cfg query_log_config.Config, parseTime query_log_tail.ParseTimeFunc) (Poller, error) {
+func New(ctx context.Context, cfg query_log_config.Config, parseTime query_log_common.ParseTimeFunc) (Poller, error) {
 	return newPoller(ctx, cfg, parseTime)
 }
 
-func newPoller(ctx context.Context, cfg query_log_config.Config, parseTime query_log_tail.ParseTimeFunc) (*poller, error) {
+func newPoller(ctx context.Context, cfg query_log_config.Config, parseTime query_log_common.ParseTimeFunc) (*poller, error) {
 	if err := cfg.Validate(); err != nil {
 		return nil, err
 	}
@@ -182,7 +182,7 @@ func (pl *poller) Commands() [][]string {
 
 func (pl *poller) TailScan(ctx context.Context, opts ...query_log_tail.OpOption) ([]Item, error) {
 	items := make([]Item, 0)
-	processMatchedFunc := func(line []byte, time time.Time, matchedFilter *query_log_filter.Filter) {
+	processMatchedFunc := func(line []byte, time time.Time, matchedFilter *query_log_common.Filter) {
 		items = append(items, Item{
 			Time:    metav1.Time{Time: time},
 			Line:    string(line),
@@ -221,7 +221,7 @@ func (pl *poller) TailScan(ctx context.Context, opts ...query_log_tail.OpOption)
 	return items, nil
 }
 
-func (pl *poller) Find(since time.Time, selectFilters ...*query_log_filter.Filter) ([]Item, error) {
+func (pl *poller) Find(since time.Time, selectFilters ...*query_log_common.Filter) ([]Item, error) {
 	// 1. filter the already flushed/in-queue ones
 	polledItems, err := pl.Poller.All(since)
 	if err != nil {
@@ -245,7 +245,7 @@ func (pl *poller) Find(since time.Time, selectFilters ...*query_log_filter.Filte
 				continue
 			}
 
-			var matchedFilter *query_log_filter.Filter
+			var matchedFilter *query_log_common.Filter
 			for _, f := range selectFilters {
 				matched, err := f.MatchString(item.Line)
 				if err != nil {
@@ -280,7 +280,7 @@ func (pl *poller) Find(since time.Time, selectFilters ...*query_log_filter.Filte
 			continue
 		}
 
-		var matchedFilter *query_log_filter.Filter
+		var matchedFilter *query_log_common.Filter
 		for _, f := range selectFilters {
 			matched, err := f.MatchString(item.Line)
 			if err != nil {

components/query/log/tail/options.go

@@ -4,7 +4,7 @@ import (
 	"errors"
 	"time"
 
-	query_log_filter "github.com/leptonai/gpud/components/query/log/filter"
+	query_log_common "github.com/leptonai/gpud/components/query/log/common"
 )
 
 type Op struct {
@@ -15,11 +15,11 @@ type Op struct {
 
 	perLineFunc func([]byte)
 
-	selectFilters []*query_log_filter.Filter
-	rejectFilters []*query_log_filter.Filter
+	selectFilters []*query_log_common.Filter
+	rejectFilters []*query_log_common.Filter
 
-	parseTime      ParseTimeFunc
-	processMatched ProcessMatchedFunc
+	parseTime      query_log_common.ParseTimeFunc
+	processMatched query_log_common.ProcessMatchedFunc
 }
 
 type OpOption func(*Op)
@@ -57,7 +57,7 @@ func (op *Op) applyOpts(opts []OpOption) error {
 		}
 	}
 	if op.processMatched == nil {
-		op.processMatched = func([]byte, time.Time, *query_log_filter.Filter) {}
+		op.processMatched = func([]byte, time.Time, *query_log_common.Filter) {}
 	}
 
 	return nil
@@ -95,7 +95,7 @@ func WithPerLineFunc(f func([]byte)) OpOption {
 // The line is sent when any of the filters match.
 // Useful for explicit blacklisting "error" logs
 // (e.g., GPU error messages in dmesg).
-func WithSelectFilter(filters ...*query_log_filter.Filter) OpOption {
+func WithSelectFilter(filters ...*query_log_common.Filter) OpOption {
 	return func(op *Op) {
 		if len(filters) > 0 {
 			op.selectFilters = append(op.selectFilters, filters...)
@@ -108,15 +108,15 @@ func WithSelectFilter(filters ...*query_log_filter.Filter) OpOption {
 // The line is sent if and only if all of the filters do not match.
 // Useful for explicit whitelisting logs and catch all other
 // (e.g., good healthy log messages).
-func WithRejectFilter(filters ...*query_log_filter.Filter) OpOption {
+func WithRejectFilter(filters ...*query_log_common.Filter) OpOption {
 	return func(op *Op) {
 		if len(filters) > 0 {
 			op.rejectFilters = append(op.rejectFilters, filters...)
 		}
 	}
 }
 
-func (op *Op) applyFilter(line any) (shouldInclude bool, matchedFilter *query_log_filter.Filter, err error) {
+func (op *Op) applyFilter(line any) (shouldInclude bool, matchedFilter *query_log_common.Filter, err error) {
 	if len(op.selectFilters) == 0 && len(op.rejectFilters) == 0 {
 		// no filters
 		return true, nil, nil
@@ -175,23 +175,19 @@ func (op *Op) applyFilter(line any) (shouldInclude bool, matchedFilter *query_lo
 	return true, matchedFilter, nil
 }
 
-type ParseTimeFunc func([]byte) (time.Time, error)
-
-func WithParseTime(f ParseTimeFunc) OpOption {
+func WithParseTime(f query_log_common.ParseTimeFunc) OpOption {
 	return func(op *Op) {
 		if f != nil {
 			op.parseTime = f
 		}
 	}
 }
 
-type ProcessMatchedFunc func([]byte, time.Time, *query_log_filter.Filter)
-
 // Called if the line is matched.
 // If not set, the matched line is no-op.
 // Useful to append to a slice or not to return a string slice
 // to avoid extra heap allocation.
-func WithProcessMatched(f ProcessMatchedFunc) OpOption {
+func WithProcessMatched(f query_log_common.ProcessMatchedFunc) OpOption {
 	return func(op *Op) {
 		if f != nil {
 			op.processMatched = f